diff --git a/fragment.go b/fragment.go index 22efb072e..e0e4743ac 100644 --- a/fragment.go +++ b/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() } diff --git a/gendebug_test.go b/gendebug_test.go new file mode 100644 index 000000000..3dd86fd8f --- /dev/null +++ b/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) +} diff --git a/generation.go b/generation.go index 34bf2b18b..7c1361b1d 100644 --- a/generation.go +++ b/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. diff --git a/generation_test.go b/generation_test.go index 3dd86fd8f..75444fcf7 100644 --- a/generation_test.go +++ b/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) -} diff --git a/logger/logger.go b/logger/logger.go index 9e895d482..3486e24a2 100644 --- a/logger/logger.go +++ b/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} +} diff --git a/mmap_test.go b/mmap_test.go new file mode 100644 index 000000000..5ce157bc4 --- /dev/null +++ b/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) + }) + } +} diff --git a/roaring/roaring.go b/roaring/roaring.go index 1ddc5f798..99a87a0d9 100644 --- a/roaring/roaring.go +++ b/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 diff --git a/roaring/unmarshal_binary.go b/roaring/unmarshal_binary.go index e80e834cf..208c15d53 100644 --- a/roaring/unmarshal_binary.go +++ b/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]) diff --git a/server/server.go b/server/server.go index e9edeb293..a8bf9e095 100644 --- a/server/server.go +++ b/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 { diff --git a/syswrap/mmap.go b/syswrap/mmap.go index 95e819999..238f6dd6f 100644 --- a/syswrap/mmap.go +++ b/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 diff --git a/syswrap/os.go b/syswrap/os.go index 1704b0759..61715f5bb 100644 --- a/syswrap/os.go +++ b/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