Merge pull request #103 from seebs/gencrash

don't mark a source as changed before we've finished remapping

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
@@ -291,18 +294,46 @@ func (f *fragment) importStorage(data []byte, file *os.File, newGen generation,
 // to use a new storage as backing store.
 func (f *fragment) applyStorage(data []byte, file *os.File, newGen generation, mapped bool) (bool, error) {
 	if len(data) == 0 {
-		return f.emptyStorage(file)
+		// This shouldn't be used anyway in this path, but just in
+		// case, we'll be explicit about it.
+		f.storage.PreferMapping(false)
+		if file != nil {
+			fi, err := file.Stat()
+			if err != nil {
+				f.Logger.Printf("trying to apply new storage to existing bitmap, stat failed: %v", err)
+			}
+			if err == nil && fi != nil && fi.Size() == 0 {
+				return f.emptyStorage(file)
+			}
+		}
+		// if we can't be sure of that, we assume data is 0 because
+		// we couldn't mmap it, and since all we'd be doing is remapping
+		// our containers to use that storage *to take advantage of
+		// mmap*, we'll just make sure our containers aren't pointing to
+		// old storage and say "nope".
+		_, _ = f.storage.RemapRoaringStorage(nil)
+		f.storage.SetSource(nil)
+		return false, nil
 	}
 	// Tell storage to prefer mapping if and only if we think the data
 	// is mmapped and valid.
 	f.storage.PreferMapping(mapped)
-	f.storage.SetSource(newGen)
 	// RemapRoaringStorage will fix any mapped containers to point either
 	// to the provided data (if PreferMapping was called with true and
 	// data is provided and there's a corresponding container) or to
 	// allocated storage, so when it's done, there's nothing in it that
 	// is mapped to anything *other than* the provided data.
-	return f.storage.RemapRoaringStorage(data)
+	mapped, err := f.storage.RemapRoaringStorage(data)
+	if err != nil {
+		// OOPS! something went wrong, we don't know why, we can't
+		// sanely recover from that.
+		_, _ = f.storage.RemapRoaringStorage(nil)
+		mapped = false
+		f.storage.SetSource(nil)
+	} else {
+		f.storage.SetSource(newGen)
+	}
+	return mapped, err
 }
 
 // openStorage opens the storage bitmap.
@@ -330,6 +361,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
@@ -1912,6 +1953,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
 }
 
@@ -2152,8 +2206,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

@@ -0,0 +1,102 @@
+// 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.
+// 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.
+
+package pilosa
+
+import (
+	"fmt"
+	"math/rand"
+	"runtime"
+	"testing"
+
+	"github.com/pilosa/pilosa/v2/logger"
+	"github.com/pilosa/pilosa/v2/syswrap"
+)
+
+type cv struct {
+	cols []uint64
+	vals []int64
+}
+
+func forceSnapshotsCheckMapping(t *testing.T) {
+	depth := uint(6)
+	f := mustOpenBSIFragment("i", "f", viewStandard, 0)
+	f.Logger = logger.NewLogfLogger(t)
+	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...
+	err := f.Snapshot()
+	if err != nil {
+		t.Fatalf("initial snapshot error: %v", err)
+	}
+
+	values := make([]cv, 1024)
+	for i := range values {
+		cols := make([]uint64, 128)
+		vals := make([]int64, 128)
+		for j := range cols {
+			// 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}
+	}
+
+	// modify the original bitmap, until it causes a snapshot, which
+	// then invalidates the other map...
+	for i := 0; i < 32; i++ {
+		cv := values[i%len(values)]
+		// periodically force gc, so if we have a small pool of maps
+		// we'll go in and out of mapping mode
+		if i%5 == 0 {
+			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)
+		}
+	}
+}
+
+// This test should basically never fail, but it might if you were running
+// out of available mmaps. Which you can fake up by adding '&& false' to the test
+// in newGeneration in generation.go. So this is probably useless but it's
+// a failure mode we've been bitten by once...
+func TestMmapBehavior(t *testing.T) {
+	var changed bool
+	var original uint64
+	defer func() {
+		syswrap.SetMaxMapCount(original)
+	}()
+
+	for _, mmapMaxVal := range []uint64{0, 3} {
+		prev := syswrap.SetMaxMapCount(mmapMaxVal)
+		if !changed {
+			original = prev
+			changed = true
+		}
+		t.Run(fmt.Sprintf("maps%d", mmapMaxVal), func(t *testing.T) {
+			forceSnapshotsCheckMapping(t)
+		})
+	}
+}

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
@@ -3752,7 +3779,10 @@ func unionArrayArrayInPlace(a, b *Container) *Container {
 			// for InPlace, we actually want to ensure that
 			// we update a, as long as it's not frozen.
 			a = a.Thaw()
-			a.setArray(b.array())
+			// ... but we also want to be sure we don't end up
+			// copying in a mapped object into our not-mapped
+			// object.
+			a.setArrayMaybeCopy(b.array(), b.Mapped())
 			return a.optimize()
 		}
 		return a

roaring/unmarshal_binary.go

@@ -88,7 +88,12 @@ func readOffsets(b *Bitmap, data []byte, pos int, keyN uint32) error {
 
 		// Map byte slice directly to the container data.
 		citer.Next()
-		_, c := citer.Value()
+		k, c := citer.Value()
+		if !c.Mapped() {
+			fmt.Printf("inexplicable: container %d (%d/%d) doesn't think it's mapped. fixing that.\n",
+				k, i, keyN)
+			c.setMapped(true)
+		}
 		switch c.typ() {
 		case containerArray:
 			c.setArray((*[0xFFFFFFF]uint16)(unsafe.Pointer(&data[offset]))[:c.N():c.N()])
@@ -108,7 +113,12 @@ func readWithRuns(b *Bitmap, data []byte, pos int, keyN uint32) error {
 	citer, _ := b.Containers.Iterator(0)
 	for i := 0; i < int(keyN); i++ {
 		citer.Next()
-		_, c := citer.Value()
+		k, c := citer.Value()
+		if !c.Mapped() {
+			fmt.Printf("inexplicable: container %d (%d/%d) doesn't think it's mapped. fixing that.\n",
+				k, i, keyN)
+			c.setMapped(true)
+		}
 		switch c.typ() {
 		case containerRun:
 			runCount := binary.LittleEndian.Uint16(data[pos : pos+runCountHeaderSize])
@@ -176,12 +186,17 @@ func (b *Bitmap) unmarshalPilosaRoaring(data []byte) error {
 
 		// Map byte slice directly to the container data.
 		citer.Next()
-		_, c := citer.Value()
+		k, c := citer.Value()
 
 		// this shouldn't happen, since we don't normally store nils.
 		if c == nil {
 			continue
 		}
+		if !c.Mapped() {
+			fmt.Printf("inexplicable: container %d (%d/%d) doesn't think it's mapped. fixing that.\n",
+				k, i, keyN)
+			c.setMapped(true)
+		}
 		switch c.typ() {
 		case containerRun:
 			runCount := binary.LittleEndian.Uint16(data[offset : offset+runCountHeaderSize])

server/server.go

@@ -229,8 +229,8 @@ func (m *Command) SetupServer() error {
 	runtime.SetBlockProfileRate(m.Config.Profile.BlockRate)
 	runtime.SetMutexProfileFraction(m.Config.Profile.MutexFraction)
 
-	syswrap.SetMaxMapCount(m.Config.MaxMapCount)
-	syswrap.SetMaxFileCount(m.Config.MaxFileCount)
+	_ = syswrap.SetMaxMapCount(m.Config.MaxMapCount)
+	_ = syswrap.SetMaxFileCount(m.Config.MaxFileCount)
 
 	err := m.setupLogger()
 	if err != nil {

syswrap/mmap.go

@@ -34,10 +34,13 @@ var ErrMaxMapCountReached = errors.New("maximum map count reached")
 var maxMapCount uint64 = 60000
 var mu sync.RWMutex
 
-func SetMaxMapCount(max uint64) {
+// SetMaxMapCount sets the maximum map count, and returns the previous maximum.
+func SetMaxMapCount(max uint64) uint64 {
 	mu.Lock()
+	prev := maxMapCount
 	maxMapCount = max
 	mu.Unlock()
+	return prev
 }
 
 // Mmap increments the global map count, and then calls syscall.Mmap. It

syswrap/os.go

@@ -27,10 +27,12 @@ var fileCount uint64
 var maxFileCount uint64 = 500000
 var fileMu sync.RWMutex
 
-func SetMaxFileCount(max uint64) {
+func SetMaxFileCount(max uint64) uint64 {
 	fileMu.Lock()
+	prev := maxFileCount
 	maxFileCount = max
 	fileMu.Unlock()
+	return prev
 }
 
 // OpenFile passes the arguments along to os.OpenFile while incrementing a