generation testing and paranoia features

We might have a problem with a stale mmap, and to try to narrow it down
a bit, we add some sanity-checking features and panic recovery to the
generation Transaction code.

This is pretty experimental.

fragment.go

@@ -28,6 +28,7 @@ import (
 	"math"
 	"math/bits"
 	"os"
+	"runtime/debug"
 	"sort"
 	"strings"
 	"sync"
@@ -105,6 +106,8 @@ type fragment struct {
 	field string
 	view  string
 	shard uint64
+	// debugging tool: addresses of current and previous maps
+	prevdata, currdata struct{ from, to uintptr }
 
 	// File-backed storage
 	path            string
@@ -352,6 +355,16 @@ func (f *fragment) openStorage(unmarshalData bool) error {
 	}
 	var err error
 	f.gen, err = newGeneration(f.gen, f.path, unmarshalData, storageOp, f.Logger)
+	if f.gen != nil {
+		scratchData := f.gen.Bytes()
+		f.prevdata = f.currdata
+		var scratchAddrs struct{ from, to uintptr }
+		if scratchData != nil {
+			scratchAddrs.from = uintptr(unsafe.Pointer(&scratchData[0]))
+			scratchAddrs.to = scratchAddrs.from + uintptr(len(scratchData))
+		}
+		f.currdata = scratchAddrs
+	}
 	if generationDebug {
 		// We might have already done this anyway, if we think we
 		// mapped stuff, but when debugging we want to do it
@@ -1934,6 +1947,19 @@ func (f *fragment) importPositions(set, clear []uint64, rowSet map[uint64]struct
 		}
 		return nil
 	})
+	if err != nil {
+		// we got an error. it's possible that the error indicates that something went wrong.
+		mappedIn, mappedOut, unmappedIn, errs, e2 := f.storage.SanityCheckMapping(f.currdata.from, f.currdata.to)
+		if errs != 0 {
+			f.Logger.Printf("transaction failed on %s. storage has %d mapped in range, %d mapped out of range, %d unmapped in range, %d errors total, last %v",
+				f.path, mappedIn, mappedOut, unmappedIn, errs, e2)
+			if f.prevdata.from != f.currdata.from {
+				mappedIn, mappedOut, unmappedIn, errs, e2 = f.storage.SanityCheckMapping(f.prevdata.from, f.prevdata.to)
+				f.Logger.Printf("with previous map, storage would have %d mapped in range, %d mapped out of range, %d unmapped in range, %d errors total, last %v",
+					mappedIn, mappedOut, unmappedIn, errs, e2)
+			}
+		}
+	}
 	return err
 }
 
@@ -2174,8 +2200,27 @@ func track(start time.Time, message string, stats stats.StatsClient, logger logg
 
 // snapshot does the actual snapshot operation. it does not check or care
 // about f.snapshotPending.
-func (f *fragment) snapshot() error {
-	_, err := unprotectedWriteToFragment(f, f.storage)
+func (f *fragment) snapshot() (err error) {
+	wouldPanic := debug.SetPanicOnFault(true)
+	defer func() {
+		debug.SetPanicOnFault(wouldPanic)
+		if r := recover(); r != nil {
+			fmt.Printf("snapshot panic!\n")
+			if e2, ok := r.(error); ok {
+				err = e2
+				// 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 e2.Error() == "runtime error: invalid memory address or nil pointer dereference" {
+					mappedIn, mappedOut, unmappedIn, errs, _ := f.storage.SanityCheckMapping(f.currdata.from, f.currdata.to)
+					f.Logger.Printf("transaction failed on %s. storage has %d mapped in range, %d mapped out of range, %d unmapped in range, %d errors total",
+						f.path, mappedIn, mappedOut, unmappedIn, errs)
+				}
+			} else {
+				err = fmt.Errorf("non-error panic: %v", r)
+			}
+		}
+	}()
+	_, err = unprotectedWriteToFragment(f, f.storage)
 	if err == nil {
 		f.snapshotStamp = time.Now()
 	}

gendebug_test.go

@@ -0,0 +1,39 @@
+// Copyright 2019 Pilosa Corp.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// +build generationdebug
+
+package pilosa
+
+import (
+	"fmt"
+	"os"
+	"testing"
+)
+
+func examineResults() {
+	results := reportGenerations()
+	if len(results) > 0 {
+		fmt.Printf("generations:\n")
+		for _, res := range results {
+			fmt.Printf("  %s\n", res)
+		}
+	}
+}
+
+func TestMain(m *testing.M) {
+	ret := m.Run()
+	examineResults()
+	os.Exit(ret)
+}

generation.go

@@ -20,6 +20,7 @@ import (
 	"io/ioutil"
 	"os"
 	"runtime"
+	"runtime/debug"
 	"sync"
 	"syscall"
 	"time"
@@ -82,7 +83,11 @@ type generation interface {
 	// ID indicates the source -- path and generation number -- that
 	// this generation represents.
 	ID() string
+	// Dead indicates whether this generation is Done.
 	Dead() bool
+	// Bytes reports the storage associated with this generation, if any.
+	// DO NOT USE THIS. Except if you're debugging mmap segfaults.
+	Bytes() []byte
 }
 
 type mmapGeneration struct {
@@ -164,9 +169,36 @@ func (m *mmapGeneration) Transaction(fileP *io.Writer, fn func() error) (transac
 	}
 	// We are done locking the generation itself for now.
 	m.mu.Unlock()
+	wouldPanic := debug.SetPanicOnFault(true)
+	defer func() {
+		debug.SetPanicOnFault(wouldPanic)
+		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" {
+					if transactionErr == nil {
+						transactionErr = errors.New("invalid memory access during transaction")
+					} else {
+						transactionErr = fmt.Errorf("invalid memory access during transaction, previous error %v", transactionErr)
+					}
+					return
+				}
+			}
+			if transactionErr == nil {
+				transactionErr = fmt.Errorf("panic during transaction: %v", r)
+			} else {
+				transactionErr = fmt.Errorf("panic during erroring transaction: panic %v, previous error %v", r, transactionErr)
+			}
+		}
+	}()
 	return fn()
 }
 
+func (m *mmapGeneration) Bytes() []byte {
+	return m.data
+}
+
 // Done marks the generation done, and closes its file, but may not unmap it.
 // It's still conceptually possible to end up doing a Transaction against a
 // done generation, but it's a red flag.

generation_test.go

@@ -1,4 +1,4 @@
-// Copyright 2019 Pilosa Corp.
+// Copyright 2020 Pilosa Corp.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -12,28 +12,56 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 //
-// +build generationdebug
+// +build generationparanoia
 
 package pilosa
 
 import (
-	"fmt"
-	"os"
+	"runtime"
 	"testing"
+	"unsafe"
 )
 
-func examineResults() {
-	results := reportGenerations()
-	if len(results) > 0 {
-		fmt.Printf("generations:\n")
-		for _, res := range results {
-			fmt.Printf("  %s\n", res)
-		}
+func TestGenerationPanic(t *testing.T) {
+	f := mustOpenFragment("i", "f", viewStandard, 0, "none")
+	defer f.Clean(t)
+
+	for i := 0; i < f.MaxOpN; i++ {
+		_, _ = f.setBit(0, uint64(i*32))
+	}
+	// force snapshot so we get a mmapped row...
+	_ = f.Snapshot()
+	_ = f.row(0)
+	var prevData []byte
+
+	if f.gen.(*mmapGeneration).data == nil {
+		t.Fatalf("generation code didn't create a mapping, apparently?")
+	}
+	prevData = f.gen.(*mmapGeneration).data
+	f.mu.Lock()
+	_ = f.snapshotQueue.Immediate(f)
+	f.mu.Unlock()
+	runtime.GC()
+	for i := 0; i < (f.MaxOpN / 2); i++ {
+		_, _ = f.setBit(0, uint64(i*32)+23)
+	}
+	f.mu.Lock()
+	f.snapshotQueue.Await(f)
+	f.mu.Unlock()
+	runtime.GC()
+	newData := f.gen.(*mmapGeneration).data
+	if unsafe.Pointer(&prevData[0]) == unsafe.Pointer(&newData[0]) {
+		t.Fatalf("test can't run usefully, didn't get new data pointer")
+	}
+
+	err := f.gen.Transaction(&f.storage.OpWriter, func() error {
+		prevData[0] = 0x3c
+		return nil
+	})
+	if err == nil {
+		t.Fatalf("expected a panic to get caught, but nothing happened")
+	}
+	if err.Error() != "invalid memory access during transaction" {
+		t.Fatalf("expected \"invalid memory access during transaction\", got %q", err.Error())
 	}
 }
-
-func TestMain(m *testing.M) {
-	ret := m.Run()
-	examineResults()
-	os.Exit(ret)
-}

logger/logger.go

@@ -105,3 +105,27 @@ func (cl *CaptureLogger) Printf(format string, v ...interface{}) {
 func (cl *CaptureLogger) Debugf(format string, v ...interface{}) {
 	cl.Debugs = append(cl.Debugs, fmt.Sprintf(format, v...))
 }
+
+// Logfer is a thing that has only a Logf() method, like for instance,
+// testing.T or testing.B.
+type Logfer interface {
+	Logf(format string, v ...interface{})
+}
+
+// LogfLogger is a logger that wraps something that has a Logf interface
+// and makes it act like our logger.
+type LogfLogger struct {
+	wrapped Logfer
+}
+
+func (ll *LogfLogger) Printf(format string, v ...interface{}) {
+	ll.wrapped.Logf(format, v...)
+}
+
+func (ll *LogfLogger) Debugf(format string, v ...interface{}) {
+	ll.wrapped.Logf(format, v...)
+}
+
+func NewLogfLogger(l Logfer) *LogfLogger {
+	return &LogfLogger{wrapped: l}
+}

mmap_test.go

@@ -16,8 +16,10 @@ package pilosa
 
 import (
 	"math/rand"
-	"sync/atomic"
+	"runtime"
 	"testing"
+
+	"github.com/pilosa/pilosa/v2/logger"
 )
 
 type cv struct {
@@ -31,37 +33,28 @@ type cv struct {
 // a failure mode we've been bitten by once...
 func TestMmapBehavior(t *testing.T) {
 	depth := uint(6)
-	var done int64
 	f := mustOpenBSIFragment("i", "f", viewStandard, 0)
+	f.Logger = logger.NewLogfLogger(t)
 	defer f.Clean(t)
 
-	ch := make(chan struct{})
-
 	for i := 0; i < f.MaxOpN; i++ {
 		_, _ = f.setBit(0, uint64(i*32))
 	}
 	// force snapshot so we get a mmapped row...
-	_ = f.Snapshot()
-	row := f.row(0)
-	segment := row.Segments()[0]
-	bitmap := segment.data
-
-	// request information from the frozen bitmap we got back
-	go func() {
-		for atomic.LoadInt64(&done) == 0 {
-			for i := 0; i < f.MaxOpN; i++ {
-				_ = bitmap.Contains(uint64(i * 32))
-			}
-		}
-		close(ch)
-	}()
+	err := f.Snapshot()
+	if err != nil {
+		t.Fatalf("initial snapshot error: %v", err)
+	}
 
 	values := make([]cv, 1024)
 	for i := range values {
-		cols := make([]uint64, 512)
-		vals := make([]int64, 512)
+		cols := make([]uint64, 128)
+		vals := make([]int64, 128)
 		for j := range cols {
-			cols[j] = uint64(rand.Int63n(ShardWidth))
+			// pick values in the first 16 cols of each of the 16
+			// shards in a default shardwidth, so each set will
+			// probably change some values from the previous one.
+			cols[j] = uint64(((rand.Int63n(16) & int64(i>>2)) << 16) + rand.Int63n(16))
 			vals[j] = int64(rand.Int63n(1 << depth))
 		}
 		values[i] = cv{cols, vals}
@@ -69,15 +62,16 @@ func TestMmapBehavior(t *testing.T) {
 
 	// modify the original bitmap, until it causes a snapshot, which
 	// then invalidates the other map...
-	for j := 0; j < 5; j++ {
-		for i := 0; i < f.MaxOpN/int(depth+1); i++ {
-			cv := values[i%len(values)]
-			err := f.importValue(cv.cols, cv.vals, depth, (i%3 == 1))
-			if err != nil {
-				t.Fatalf("importValue[%d][%d]: %v", j, i, err)
-			}
+	for i := 0; i < 32; i++ {
+		cv := values[i%len(values)]
+		runtime.GC()
+		err := f.importValue(cv.cols, cv.vals, depth, (i%3 == 1))
+		if err != nil {
+			t.Fatalf("importValue[%d]: %v", i, err)
+		}
+		err = f.Snapshot()
+		if err != nil {
+			t.Fatalf("snapshot[%d]: %v", i, err)
 		}
 	}
-	atomic.StoreInt64(&done, 1)
-	<-ch
 }

roaring/roaring.go

@@ -1529,6 +1529,33 @@ func (r *baseRoaringIterator) Current() (key uint64, cType byte, n int, length i
 	return r.currentKey, r.currentType, r.currentN, r.currentLen, r.currentPointer, r.lastErr
 }
 
+// SanityCheckMapping is a debugging function which checks whether containers
+// are *correctly* recorded as mapped or unmapped.
+func (b *Bitmap) SanityCheckMapping(from, to uintptr) (mappedIn int64, mappedOut int64, unmappedIn int64, errs int, err error) {
+	b.Containers.UpdateEvery(func(key uint64, c *Container, existed bool) (*Container, bool) {
+		dptr := uintptr(unsafe.Pointer(c.pointer))
+		if dptr >= from && dptr < to {
+			if c.Mapped() {
+				mappedIn++
+			} else {
+				err = fmt.Errorf("container key %d, addr %x, inside %x+%d\n",
+					key, dptr, from, to-from)
+				errs++
+				unmappedIn++
+			}
+		} else {
+			if c.Mapped() {
+				err = fmt.Errorf("container key %d, addr %x, outside %x+%d, but mapped\n",
+					key, dptr, from, to-from)
+				errs++
+				mappedOut++
+			}
+		}
+		return c, false
+	})
+	return mappedIn, mappedOut, unmappedIn, errs, err
+}
+
 // RemapRoaringStorage tries to update all containers to refer to
 // the roaring bitmap in the provided []byte. If any containers are
 // marked as mapped, but do not match the provided storage, they will