diff --git a/cmd/root.go b/cmd/root.go index 3ff5c22e4..6d0ad9ee2 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -75,6 +75,7 @@ Build Time: ` + pilosa.BuildTime + "\n", rc.AddCommand(newImportCommand(stdin, stdout, stderr)) rc.AddCommand(newInspectCommand(stdin, stdout, stderr)) rc.AddCommand(newServeCmd(stdin, stdout, stderr)) + rc.AddCommand(newHolderCmd(stdin, stdout, stderr)) rc.SetOutput(stderr) return rc diff --git a/cmd/server.go b/cmd/server.go index 43432fceb..f57678d52 100644 --- a/cmd/server.go +++ b/cmd/server.go @@ -28,6 +28,33 @@ import ( // Server is global so that tests can control and verify it. var Server *server.Command +var holder *server.Command + +// newHolderCmd creates a pilosa server for just long enough to open the +// holder, then shuts it down again. +func newHolderCmd(stdin io.Reader, stdout, stderr io.Writer) *cobra.Command { + holder = server.NewCommand(stdin, stdout, stderr) + serveCmd := &cobra.Command{ + Use: "holder", + Short: "Load Pilosa.", + Long: `pilosa holder starts (and immediately stops) Pilosa. + +It opens the data directory and loads it, then shuts down immediately. +This is only useful for diagnostic use. +`, + RunE: func(cmd *cobra.Command, args []string) error { + // Start & run the server. + if err := holder.UpAndDown(); err != nil { + return errors.Wrap(err, "running server") + } + return nil + }, + } + + // Attach flags to the command. + ctl.BuildServerFlags(serveCmd, holder) + return serveCmd +} // newServeCmd creates a pilosa server and runs it with command line flags. func newServeCmd(stdin io.Reader, stdout, stderr io.Writer) *cobra.Command { diff --git a/ctl/check_test.go b/ctl/check_test.go index e37d99358..ab5ef9279 100644 --- a/ctl/check_test.go +++ b/ctl/check_test.go @@ -93,8 +93,9 @@ func TestCheckCommand_Run(t *testing.T) { t.Fatalf("copy: %v", err) } - if !strings.HasPrefix(err.Error(), "checking bitmap: unmarshalling: reading roaring header:") { - t.Fatalf("expect error: invalid roaring file, actual: '%s'", err) + expectedPrefix := "checking bitmap: unmarshalling: " + if !strings.HasPrefix(err.Error(), expectedPrefix) { + t.Fatalf("expect error: '%s...', actual: '%s'", expectedPrefix, err) } // Todo: need correct roaring file for happy path } diff --git a/ctl/inspect_test.go b/ctl/inspect_test.go index bb87f894d..1ff852712 100644 --- a/ctl/inspect_test.go +++ b/ctl/inspect_test.go @@ -41,8 +41,9 @@ func TestInspectCommand_Run(t *testing.T) { file.Close() cm.Path = file.Name() err = cm.Run(context.Background()) - if err != nil && err.Error() != "unmarshalling: reading roaring header: did not find expected serialCookie in header" { - t.Fatalf("can't run command: %v", err) + expectedError := "unmarshalling: " + if !strings.Contains(err.Error(), expectedError) { + t.Fatalf("expected error '%s', got '%v'", expectedError, err) } w.Close() diff --git a/field.go b/field.go index c803ea36c..39c6c5363 100644 --- a/field.go +++ b/field.go @@ -35,6 +35,7 @@ import ( "github.com/pilosa/pilosa/stats" "github.com/pilosa/pilosa/tracing" "github.com/pkg/errors" + "golang.org/x/sync/errgroup" ) // Default field settings. @@ -432,6 +433,8 @@ func (f *Field) Open() error { return nil } +var fieldQueue = make(chan struct{}, 16) + // openViews opens and initializes the views inside the field. func (f *Field) openViews() error { file, err := os.Open(filepath.Join(f.path, "views")) @@ -446,40 +449,56 @@ func (f *Field) openViews() error { if err != nil { return errors.Wrap(err, "reading directory") } + eg, ctx := errgroup.WithContext(context.Background()) + var mu sync.Mutex - for _, fi := range fis { - if !fi.IsDir() { - continue - } - - name := filepath.Base(fi.Name()) - f.logger.Debugf("open index/field/view: %s/%s/%s", f.index, f.name, fi.Name()) - view := f.newView(f.viewPath(name), name) - if err := view.open(); err != nil { - return fmt.Errorf("opening view: view=%s, err=%s", view.name, err) - } - - // Automatically upgrade BSI v1 fragments if they exist & reopen view. - if bsig := f.bsiGroup(f.name); bsig != nil { - if ok, err := upgradeViewBSIv2(view, bsig.BitDepth); err != nil { - return errors.Wrap(err, "upgrade view bsi v2") - } else if ok { - if err := view.close(); err != nil { - return errors.Wrap(err, "closing upgraded view") - } - view = f.newView(f.viewPath(name), name) - if err := view.open(); err != nil { - return fmt.Errorf("re-opening view: view=%s, err=%s", view.name, err) - } + for _, loopFi := range fis { + select { + case <-ctx.Done(): + break + default: + fi := loopFi + if !fi.IsDir() { + continue } - } + fieldQueue <- struct{}{} + eg.Go(func() error { + defer func() { + <-fieldQueue + }() + name := filepath.Base(fi.Name()) + f.logger.Debugf("open index/field/view: %s/%s/%s", f.index, f.name, fi.Name()) + view := f.newView(f.viewPath(name), name) + if err := view.open(); err != nil { + return fmt.Errorf("opening view: view=%s, err=%s", view.name, err) + } - view.rowAttrStore = f.rowAttrStore - f.logger.Debugf("add index/field/view to field.viewMap: %s/%s/%s", f.index, f.name, view.name) - f.viewMap[view.name] = view + // Automatically upgrade BSI v1 fragments if they exist & reopen view. + if bsig := f.bsiGroup(f.name); bsig != nil { + if ok, err := upgradeViewBSIv2(view, bsig.BitDepth); err != nil { + return errors.Wrap(err, "upgrade view bsi v2") + } else if ok { + if err := view.close(); err != nil { + return errors.Wrap(err, "closing upgraded view") + } + view = f.newView(f.viewPath(name), name) + if err := view.open(); err != nil { + return fmt.Errorf("re-opening view: view=%s, err=%s", view.name, err) + } + } + } + + view.rowAttrStore = f.rowAttrStore + f.logger.Debugf("add index/field/view to field.viewMap: %s/%s/%s", f.index, f.name, view.name) + mu.Lock() + f.viewMap[view.name] = view + mu.Unlock() + return nil + }) + } } - return nil + return eg.Wait() } // loadMeta reads meta data for the field, if any. diff --git a/fragment.go b/fragment.go index 94b86ac8a..ad43b2503 100644 --- a/fragment.go +++ b/fragment.go @@ -399,6 +399,8 @@ func (f *fragment) openStorage(unmarshalData bool) error { } }() } + // set the preference for mapping based on whether the data's mmapped + f.storage.PreferMapping(newStorageData != nil) // so we have a problem here: if this fails, it's unclear whether // *either* or *both* of old and new storage data might be in use. // So we call the thing that should unconditionally unmap both of them... diff --git a/index.go b/index.go index bef25c9c5..e02557501 100644 --- a/index.go +++ b/index.go @@ -15,6 +15,7 @@ package pilosa import ( + "context" "fmt" "io/ioutil" "os" @@ -29,6 +30,7 @@ import ( "github.com/pilosa/pilosa/roaring" "github.com/pilosa/pilosa/stats" "github.com/pkg/errors" + "golang.org/x/sync/errgroup" ) // Index represents a container for fields. @@ -137,6 +139,8 @@ func (i *Index) Open() error { return nil } +var indexQueue = make(chan struct{}, 8) + // openFields opens and initializes the fields inside the index. func (i *Index) openFields() error { f, err := os.Open(i.path) @@ -149,24 +153,43 @@ func (i *Index) openFields() error { if err != nil { return errors.Wrap(err, "reading directory") } + eg, ctx := errgroup.WithContext(context.Background()) + var mu sync.Mutex - for _, fi := range fis { - if !fi.IsDir() { - continue - } + for _, loopFi := range fis { + select { + case <-ctx.Done(): + break + default: + fi := loopFi + if !fi.IsDir() { + continue + } + indexQueue <- struct{}{} + eg.Go(func() error { + defer func() { + <-indexQueue + }() + i.logger.Debugf("open field: %s", fi.Name()) + mu.Lock() + fld, err := i.newField(i.fieldPath(filepath.Base(fi.Name())), filepath.Base(fi.Name())) + mu.Unlock() + if err != nil { + return errors.Wrapf(ErrName, "'%s'", fi.Name()) + } - i.logger.Debugf("open field: %s", fi.Name()) - fld, err := i.newField(i.fieldPath(filepath.Base(fi.Name())), filepath.Base(fi.Name())) - if err != nil { - return errors.Wrapf(ErrName, "'%s'", fi.Name()) + if err := fld.Open(); err != nil { + return fmt.Errorf("open field: name=%s, err=%s", fld.Name(), err) + } + i.logger.Debugf("add field to index.fields: %s", fi.Name()) + mu.Lock() + i.fields[fld.Name()] = fld + mu.Unlock() + return nil + }) } - if err := fld.Open(); err != nil { - return fmt.Errorf("open field: name=%s, err=%s", fld.Name(), err) - } - i.logger.Debugf("add field to index.fields: %s", fi.Name()) - i.fields[fld.Name()] = fld } - return nil + return eg.Wait() } // openExistenceField gets or creates the existence field and associates it to the index. diff --git a/roaring/fuzz_test.go b/roaring/fuzz_test.go index fe8fb6cee..86f99faa0 100644 --- a/roaring/fuzz_test.go +++ b/roaring/fuzz_test.go @@ -14,6 +14,7 @@ package roaring import ( + "strings" "testing" ) @@ -25,53 +26,50 @@ func TestUnmarshalBinary(t *testing.T) { }{ { // Checks for the zero containers situation cr: []byte(":0\x00\x00\x01\x00\x00\x000000"), //":000000" - expected: "reading roaring header: malformed bitmap, key-cardinality slice overruns buffer at 12", - }, - { // Checks for int overflow - cr: []byte("<0\x000\x00\x00\x00\x00000000000000" + - "0"), //"<000000000000000" - expected: "unmarshaling as pilosa roaring: unknown op type: 48", + expected: "header: malformed bitmap, key-cardinality slice overruns buffer at 12", }, { // The next 5 check for malformed bitmaps cr: []byte("<0\x0000000000000000000" + "\x00\x00\xec\x00\x03\x00\x00\x00\xec000"), //"<000000000000000000ÏÏ000" - expected: "unmarshaling as pilosa roaring: malformed bitmap, key-cardinality not provided for 67372036 containers", + expected: "insufficient data for header + offsets:", }, { cr: []byte("<0\x00\x02\x00\x00\x00\\f\x01\xb5\x8d\x009\v\x01\x00\x00\x00\x00" + "\x00\x00e\x04\x00\x00\x00\x04\xfd\x00\x01\x00"), //"<0\fµç9e˝" - expected: "unmarshaling as pilosa roaring: malformed bitmap, key-cardinality not provided for 128625322 containers", + expected: "insufficient data for header + offsets:", }, { cr: []byte("<0\x00\x02\x00\x00\x00&x.field safe"), //"<0&x.field safe" - expected: "unmarshaling as pilosa roaring: malformed bitmap, key-cardinality not provided for 53127850 containers", + expected: "insufficient data for header + offsets:", }, { cr: []byte("<0\x00\x00\x14\x00\x00\x00\x80\xffp\x05_ 4\x114089" + "\x00\x00\xff\x000\x00\x02\x00\x00\x00\x00\xff\u007f\x00\x00\x01\x10\x00\x00j" + "\x02\x00\x00$\x04_\x00\xff\u007f\xff062616163\x00" + //"<0ġp_ 44089ˇ0ˇj$_ˇˇ0626161630ø¸ad$j√" "0\x00\x02\x00\x01\xbf\x00\x04\x00\xfcad$\x00\x00j\x10\x00\x00\xc3"), - expected: "unmarshaling as pilosa roaring: malformed bitmap, key-cardinality not provided for 1 containers", + expected: "insufficient data for header + offsets:", }, { // 0 containers because the container is partially formed, but not fully (ie. 3/12 = 0) cr: []byte("<0\x00\x02\x03\x00\x00\x00쳫\v\x00d9\v\x00\x009\v"), //<0쳫 d9 9 - expected: "unmarshaling as pilosa roaring: malformed bitmap, key-cardinality not provided for 0 containers", + expected: "insufficient data for header + offsets:", }, { // Checks for incomplete offset in readWithRuns cr: []byte(";0\x00\x00\v00000"), //";00 00000" - expected: "reading offsets from official roaring format: offset incomplete: len=10", + expected: "insufficient data for offsets", }, { // Checks for incomplete offset in readOffsets cr: []byte(":0\x00\x00\x03\x00\x00\x00000000000000" + "\x00"), //:0000000000000 - expected: "reading offsets from official roaring format: offset incomplete: len=1", + expected: "insufficient data for offsets", }, } for _, crash := range confirmedCrashers { err := b.UnmarshalBinary(crash.cr) - if err.Error() != crash.expected { - t.Errorf("Expected: %s, Got: %s", crash.expected, err) + if err == nil { + t.Errorf("expected: %s, got: no error", crash.expected) + } else if !strings.Contains(err.Error(), crash.expected) { + t.Errorf("expected: %s, got: %s", crash.expected, err) } } diff --git a/roaring/roaring.go b/roaring/roaring.go index 92ca8bb18..86ee64fc5 100644 --- a/roaring/roaring.go +++ b/roaring/roaring.go @@ -147,6 +147,8 @@ type Bitmap struct { // User-defined flags. Flags byte + // should we try to keep things mapped? + preferMapping bool // Number of bit change operations written to the writer. Some operations // contain multiple values, so "ops" represents the number of distinct @@ -1125,7 +1127,11 @@ func (b *Bitmap) writeToUnoptimized(w io.Writer) (n int64, err error) { // bitmap and yield information about containers, including type, size, and // the location of their data structures. type roaringIterator interface { + // Next yields the information about the next container Next() (key uint64, cType byte, n int, length int, pointer *uint16, err error) + // Remaining yields the bytes left over past the end of the roaring data, + // which is typically an ops log in our case. + Remaining() []byte } // baseRoaringIterator holds values used by both Pilosa and official Roaring @@ -1142,6 +1148,7 @@ type baseRoaringIterator struct { currentLen int currentPointer *uint16 currentDataOffset uint32 + lastDataOffset int64 lastErr error } @@ -1189,10 +1196,14 @@ func newOfficialRoaringIterator(data []byte) (*officialRoaringIterator, error) { r.headers = data[headerOffset:offsetOffset] // note: offsets are only actually used with the no-run headers. if r.haveRuns { - // start out pointed at where the offsets would have been. r.currentDataOffset = uint32(offsetOffset) } else { + if len(r.data) < offsetOffset+int(r.keys*4) { + return nil, fmt.Errorf("insufficient data for offsets (need %d bytes, found %d)", + r.keys*4, len(r.data)-offsetOffset) + } r.offsets = data[offsetOffset : offsetOffset+int(r.keys*4)] + r.currentDataOffset = uint32(offsetOffset) } // set key to -1; user should call Next first. r.currentIdx = -1 @@ -1212,6 +1223,12 @@ func newPilosaRoaringIterator(data []byte) (*pilosaRoaringIterator, error) { r.keys = int64(binary.LittleEndian.Uint32(data[3+1 : 8])) // it could happen if r.keys == 0 { + // special case: what if we have zero containers, but a valid ops log after them? + // set currentDataOffset so that Done will set lastDataOffset and Remaining() will + // work. + if len(data) > headerBaseSize { + r.currentDataOffset = headerBaseSize + } // not an error, exactly. it's valid and well-formed, we just have nothing to do r.Done(io.EOF) return r, nil @@ -1227,6 +1244,10 @@ func newPilosaRoaringIterator(data []byte) (*pilosaRoaringIterator, error) { offsetEnd := offsetStart + (r.keys * 4) r.headers = data[headerStart:headerEnd] r.offsets = data[offsetStart:offsetEnd] + // if there's no containers, we want to act as though data started at the end + // of the list of offsets, which was also empty, so we don't think the entire thing + // is actually a malformed op + r.currentDataOffset = uint32(offsetEnd) // set key to -1; user should call Next first. r.currentIdx = -1 r.currentKey = ^uint64(0) @@ -1257,9 +1278,17 @@ func (r *baseRoaringIterator) Done(err error) { r.currentN = 0 r.currentLen = 0 r.currentPointer = nil + r.lastDataOffset = int64(r.currentDataOffset) r.currentDataOffset = 0 } +func (r *baseRoaringIterator) Remaining() []byte { + if r.lastDataOffset == 0 { + return nil + } + return r.data[r.lastDataOffset:] +} + func (r *pilosaRoaringIterator) Next() (key uint64, cType byte, n int, length int, pointer *uint16, err error) { if r.currentIdx >= r.keys { // we're already done @@ -1306,6 +1335,7 @@ func (r *pilosaRoaringIterator) Next() (key uint64, cType byte, n int, length in r.currentIdx, r.keys, r.currentKey, r.currentDataOffset, size, len(r.data))) return r.Current() } + r.currentDataOffset += uint32(size) r.lastErr = nil return r.Current() } @@ -1333,6 +1363,11 @@ func (r *officialRoaringIterator) Next() (key uint64, cType byte, n int, length // a run container keeps its data after an initial 2 byte length header var runCount uint16 if r.currentType == containerRun { + if int(r.currentDataOffset)+2 > len(r.data) { + r.Done(fmt.Errorf("insufficient data for offsets container %d/%d, expect run length at %d/%d bytes", + r.currentIdx, r.keys, r.currentDataOffset, len(r.data))) + return r.Current() + } runCount = binary.LittleEndian.Uint16(r.data[r.currentDataOffset : r.currentDataOffset+runCountHeaderSize]) r.currentDataOffset += 2 } @@ -1554,102 +1589,10 @@ func (b *Bitmap) ImportRoaringBits(data []byte, clear bool, log bool, rowSize ui err = b.writeOp(&op) } return changed, rowSet, err - } -// unmarshalPilosaRoaring treats data as being encoded in Pilosa's 64 bit -// roaring format and decodes it into b. -func (b *Bitmap) unmarshalPilosaRoaring(data []byte) error { - if len(data) < headerBaseSize { - return errors.New("data too small") - } - - // Verify the first two bytes are a valid MagicNumber, and second two bytes match current storageVersion. - fileMagic := uint32(binary.LittleEndian.Uint16(data[0:2])) - fileVersion := uint32(data[2]) - b.Flags = data[3] - if fileMagic != MagicNumber { - return fmt.Errorf("invalid roaring file, magic number %v is incorrect", fileMagic) - } - - if fileVersion != storageVersion { - return fmt.Errorf("wrong roaring version, file is v%d, server requires v%d", fileVersion, storageVersion) - } - - // Read key count in bytes sizeof(cookie)+sizeof(flag):(sizeof(cookie)+sizeof(uint32)). - keyN := binary.LittleEndian.Uint32(data[3+1 : 8]) - if uint32(len(data)) < headerBaseSize+keyN*12 { - return fmt.Errorf("malformed bitmap, key-cardinality not provided for %d containers", int(keyN)/12) - } - - headerSize := headerBaseSize - b.Containers.ResetN(int(keyN)) - // Descriptive header section: Read container keys and cardinalities. - for i, buf := 0, data[headerSize:]; i < int(keyN); i, buf = i+1, buf[12:] { - b.Containers.PutContainerValues( - binary.LittleEndian.Uint64(buf[0:8]), - byte(binary.LittleEndian.Uint16(buf[8:10])), - int(binary.LittleEndian.Uint16(buf[10:12]))+1, - true) - } - opsOffset := headerSize + int(keyN)*12 - - // Read container offsets and attach data. - citer, _ := b.Containers.Iterator(0) - for i, buf := 0, data[opsOffset:]; i < int(keyN); i, buf = i+1, buf[4:] { - offset := binary.LittleEndian.Uint32(buf[0:4]) - // Verify the offset is within the bounds of the input data. - if int(offset) >= len(data) { - return fmt.Errorf("offset out of bounds: off=%d, len=%d", offset, len(data)) - } - - // Map byte slice directly to the container data. - citer.Next() - _, c := citer.Value() - // this shouldn't happen, since we don't normally store nils. - if c == nil { - continue - } - switch c.typ() { - case containerRun: - runCount := binary.LittleEndian.Uint16(data[offset : offset+runCountHeaderSize]) - c.setRuns((*[0xFFFFFFF]interval16)(unsafe.Pointer(&data[offset+runCountHeaderSize]))[:runCount:runCount]) - opsOffset = int(offset) + runCountHeaderSize + len(c.runs())*interval16Size - case containerArray: - c.setArray((*[0xFFFFFFF]uint16)(unsafe.Pointer(&data[offset]))[:c.N():c.N()]) - opsOffset = int(offset) + len(c.array())*2 // sizeof(uint32) - case containerBitmap: - c.setBitmap((*[0xFFFFFFF]uint64)(unsafe.Pointer(&data[offset]))[:bitmapN:bitmapN]) - opsOffset = int(offset) + len(c.bitmap())*8 // sizeof(uint64) - } - } - - // Read ops log until the end of the file. - buf := data[opsOffset:] - for { - // Exit when there are no more ops to parse. - if len(buf) == 0 { - break - } - - // Unmarshal the op and apply it. - var opr op - if err := opr.UnmarshalBinary(buf); err != nil { - // FIXME(benbjohnson): return error with position so file can be trimmed. - return err - } - - opr.apply(b) - - // Increase the op count. - b.ops++ - b.opN += opr.count() - - // Move the buffer forward. - buf = buf[opr.size():] - } - - return nil +func (b *Bitmap) PreferMapping(preferred bool) { + b.preferMapping = preferred } // writeOp writes op to the OpWriter, if available. @@ -5138,112 +5081,6 @@ func readOfficialHeader(buf []byte) (size uint32, containerTyper func(index uint return size, containerTyper, header, pos, haveRuns, err } -// UnmarshalBinary decodes b from a binary-encoded byte slice. data can be in -// either official roaring format or Pilosa's roaring format. -func (b *Bitmap) UnmarshalBinary(data []byte) error { - if data == nil { - // Nothing to unmarshal - return nil - } - statsHit("Bitmap/UnmarshalBinary") - b.opN = 0 // reset opN since we're reading new data. - fileMagic := uint32(binary.LittleEndian.Uint16(data[0:2])) - if fileMagic == MagicNumber { // if pilosa roaring - return errors.Wrap(b.unmarshalPilosaRoaring(data), "unmarshaling as pilosa roaring") - } - - keyN, containerTyper, header, pos, haveRuns, err := readOfficialHeader(data) - if err != nil { - return errors.Wrap(err, "reading roaring header") - } - // Only the Pilosa roaring format has flags. The official Roaring format - // hasn't got space in its header for flags. - b.Flags = 0 - - b.Containers.ResetN(int(keyN)) - // Descriptive header section: Read container keys and cardinalities. - for i, buf := uint(0), data[header:]; i < uint(keyN); i, buf = i+1, buf[4:] { - card := int(binary.LittleEndian.Uint16(buf[2:4])) + 1 - b.Containers.PutContainerValues( - uint64(binary.LittleEndian.Uint16(buf[0:2])), - containerTyper(i, card), /// container type voodo with isRunBitmap - card, - true) - } - - // Read container offsets and attach data. - if haveRuns { - err := readWithRuns(b, data, pos, keyN) - if err != nil { - return errors.Wrap(err, "reading offsets from official roaring format") - } - } else { - err := readOffsets(b, data, pos, keyN) - if err != nil { - return errors.Wrap(err, "reading offsets from official roaring format") - } - } - return nil -} - -func readOffsets(b *Bitmap, data []byte, pos int, keyN uint32) error { - - citer, _ := b.Containers.Iterator(0) - for i, buf := 0, data[pos:]; i < int(keyN); i, buf = i+1, buf[4:] { - // Verify the offset is fully formed - if len(buf) < 4 { - return fmt.Errorf("offset incomplete: len=%d", len(buf)) - } - offset := binary.LittleEndian.Uint32(buf[0:4]) - // Verify the offset is within the bounds of the input data. - if int(offset) >= len(data) { - return fmt.Errorf("offset out of bounds: off=%d, len=%d", offset, len(data)) - } - - // Map byte slice directly to the container data. - citer.Next() - _, c := citer.Value() - switch c.typ() { - case containerArray: - c.setArray((*[0xFFFFFFF]uint16)(unsafe.Pointer(&data[offset]))[:c.N():c.N()]) - case containerBitmap: - c.setBitmap((*[0xFFFFFFF]uint64)(unsafe.Pointer(&data[offset]))[:bitmapN:bitmapN]) - default: - return fmt.Errorf("unsupported container type %d", c.typ()) - } - } - return nil -} - -func readWithRuns(b *Bitmap, data []byte, pos int, keyN uint32) error { - if len(data) < pos+runCountHeaderSize { - return fmt.Errorf("offset incomplete: len=%d", len(data)) - } - citer, _ := b.Containers.Iterator(0) - for i := 0; i < int(keyN); i++ { - citer.Next() - _, c := citer.Value() - switch c.typ() { - case containerRun: - runCount := binary.LittleEndian.Uint16(data[pos : pos+runCountHeaderSize]) - c.setRuns((*[0xFFFFFFF]interval16)(unsafe.Pointer(&data[pos+runCountHeaderSize]))[:runCount:runCount]) - runs := c.runs() - - for o := range runs { // must convert from start:length to start:end :( - runs[o].last = runs[o].start + runs[o].last - } - pos += int((runCount * interval16Size) + runCountHeaderSize) - case containerArray: - c.setArray((*[0xFFFFFFF]uint16)(unsafe.Pointer(&data[pos]))[:c.N():c.N()]) - pos += int(c.N() * 2) - case containerBitmap: - c.setBitmap((*[0xFFFFFFF]uint64)(unsafe.Pointer(&data[pos]))[:bitmapN:bitmapN]) - pos += bitmapN * 8 - } - } - return nil -} - // handledIter and handledIters are wrappers around Bitmap Container iterators // and assist with the unionIntoTarget algorithm by abstracting away some tedious // operations. diff --git a/roaring/roaring_internal_test.go b/roaring/roaring_internal_test.go index 22b0c3f03..0ecd73021 100644 --- a/roaring/roaring_internal_test.go +++ b/roaring/roaring_internal_test.go @@ -3409,11 +3409,11 @@ func TestUnmarshalRoaringWithErrors(t *testing.T) { }{ { // Runs a bitmap without runs and no containers through the official roaring hexString: "3A30000000000000", - expectedError: "reading roaring header: malformed bitmap, key-cardinality slice overruns buffer at 8", + expectedError: "header: malformed bitmap, key-cardinality slice overruns buffer at 8", }, { // Runs a bitmap with runs and no containers through the official roaring hexString: "3B30000000000000", - expectedError: "reading roaring header: malformed bitmap, key-cardinality slice overruns buffer at 9", + expectedError: "header: malformed bitmap, key-cardinality slice overruns buffer at 9", }, { // Runs a bitmap in the Pilosa format through the Pilosa roaring hexString: "3C30000000000000", @@ -3427,7 +3427,7 @@ func TestUnmarshalRoaringWithErrors(t *testing.T) { bm := NewBitmap() err = bm.UnmarshalBinary(zeroContainers) if err != nil { - if err.Error() != loopContainers.expectedError { + if !strings.Contains(err.Error(), loopContainers.expectedError) { t.Fatalf("Expected: %s, Got: %s", loopContainers.expectedError, err) } } diff --git a/roaring/unmarshal_binary.go b/roaring/unmarshal_binary.go new file mode 100644 index 000000000..6b13cce0f --- /dev/null +++ b/roaring/unmarshal_binary.go @@ -0,0 +1,222 @@ +// Copyright 2017 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 !enterprise + +package roaring + +import ( + "encoding/binary" + "fmt" + "unsafe" + + "github.com/pkg/errors" +) + +// UnmarshalBinary decodes b from a binary-encoded byte slice. data can be in +// either official roaring format or Pilosa's roaring format. +func (b *Bitmap) UnmarshalBinary(data []byte) error { + if data == nil { + // Nothing to unmarshal + return nil + } + statsHit("Bitmap/UnmarshalBinary") + b.opN = 0 // reset opN since we're reading new data. + fileMagic := uint32(binary.LittleEndian.Uint16(data[0:2])) + if fileMagic == MagicNumber { // if pilosa roaring + return errors.Wrap(b.unmarshalPilosaRoaring(data), "unmarshaling as pilosa roaring") + } + + keyN, containerTyper, header, pos, haveRuns, err := readOfficialHeader(data) + if err != nil { + return errors.Wrap(err, "reading roaring header") + } + // Only the Pilosa roaring format has flags. The official Roaring format + // hasn't got space in its header for flags. + b.Flags = 0 + + b.Containers.ResetN(int(keyN)) + // Descriptive header section: Read container keys and cardinalities. + for i, buf := uint(0), data[header:]; i < uint(keyN); i, buf = i+1, buf[4:] { + card := int(binary.LittleEndian.Uint16(buf[2:4])) + 1 + b.Containers.PutContainerValues( + uint64(binary.LittleEndian.Uint16(buf[0:2])), + containerTyper(i, card), /// container type voodo with isRunBitmap + card, + true) + } + + // Read container offsets and attach data. + if haveRuns { + err := readWithRuns(b, data, pos, keyN) + if err != nil { + return errors.Wrap(err, "reading offsets from official roaring format") + } + } else { + err := readOffsets(b, data, pos, keyN) + if err != nil { + return errors.Wrap(err, "reading official roaring format") + } + } + return nil +} + +func readOffsets(b *Bitmap, data []byte, pos int, keyN uint32) error { + + citer, _ := b.Containers.Iterator(0) + for i, buf := 0, data[pos:]; i < int(keyN); i, buf = i+1, buf[4:] { + // Verify the offset is fully formed + if len(buf) < 4 { + return fmt.Errorf("insufficient data for offsets: len=%d", len(buf)) + } + offset := binary.LittleEndian.Uint32(buf[0:4]) + // Verify the offset is within the bounds of the input data. + if int(offset) >= len(data) { + return fmt.Errorf("offset out of bounds: off=%d, len=%d", offset, len(data)) + } + + // Map byte slice directly to the container data. + citer.Next() + _, c := citer.Value() + switch c.typ() { + case containerArray: + c.setArray((*[0xFFFFFFF]uint16)(unsafe.Pointer(&data[offset]))[:c.N():c.N()]) + case containerBitmap: + c.setBitmap((*[0xFFFFFFF]uint64)(unsafe.Pointer(&data[offset]))[:bitmapN:bitmapN]) + default: + return fmt.Errorf("unsupported container type %d", c.typ()) + } + } + return nil +} + +func readWithRuns(b *Bitmap, data []byte, pos int, keyN uint32) error { + if len(data) < pos+runCountHeaderSize { + return fmt.Errorf("insufficient data for offsets(run): len=%d", len(data)) + } + citer, _ := b.Containers.Iterator(0) + for i := 0; i < int(keyN); i++ { + citer.Next() + _, c := citer.Value() + switch c.typ() { + case containerRun: + runCount := binary.LittleEndian.Uint16(data[pos : pos+runCountHeaderSize]) + c.setRuns((*[0xFFFFFFF]interval16)(unsafe.Pointer(&data[pos+runCountHeaderSize]))[:runCount:runCount]) + runs := c.runs() + + for o := range runs { // must convert from start:length to start:end :( + runs[o].last = runs[o].start + runs[o].last + } + pos += int((runCount * interval16Size) + runCountHeaderSize) + case containerArray: + c.setArray((*[0xFFFFFFF]uint16)(unsafe.Pointer(&data[pos]))[:c.N():c.N()]) + pos += int(c.N() * 2) + case containerBitmap: + c.setBitmap((*[0xFFFFFFF]uint64)(unsafe.Pointer(&data[pos]))[:bitmapN:bitmapN]) + pos += bitmapN * 8 + } + } + return nil +} + +func (b *Bitmap) unmarshalPilosaRoaring(data []byte) error { + if len(data) < headerBaseSize { + return errors.New("data too small") + } + + // Verify the first two bytes are a valid MagicNumber, and second two bytes match current storageVersion. + fileMagic := uint32(binary.LittleEndian.Uint16(data[0:2])) + fileVersion := uint32(data[2]) + b.Flags = data[3] + if fileMagic != MagicNumber { + return fmt.Errorf("invalid roaring file, magic number %v is incorrect", fileMagic) + } + + if fileVersion != storageVersion { + return fmt.Errorf("wrong roaring version, file is v%d, server requires v%d", fileVersion, storageVersion) + } + + // Read key count in bytes sizeof(cookie)+sizeof(flag):(sizeof(cookie)+sizeof(uint32)). + keyN := binary.LittleEndian.Uint32(data[3+1 : 8]) + if uint32(len(data)) < headerBaseSize+keyN*12 { + return fmt.Errorf("insufficient data for header + offsets: key-cardinality not provided for %d containers", int(keyN)/12) + } + + headerSize := headerBaseSize + b.Containers.ResetN(int(keyN)) + // Descriptive header section: Read container keys and cardinalities. + for i, buf := 0, data[headerSize:]; i < int(keyN); i, buf = i+1, buf[12:] { + b.Containers.PutContainerValues( + binary.LittleEndian.Uint64(buf[0:8]), + byte(binary.LittleEndian.Uint16(buf[8:10])), + int(binary.LittleEndian.Uint16(buf[10:12]))+1, + true) + } + opsOffset := headerSize + int(keyN)*12 + + // Read container offsets and attach data. + citer, _ := b.Containers.Iterator(0) + for i, buf := 0, data[opsOffset:]; i < int(keyN); i, buf = i+1, buf[4:] { + offset := binary.LittleEndian.Uint32(buf[0:4]) + // Verify the offset is within the bounds of the input data. + if int(offset) >= len(data) { + return fmt.Errorf("offset out of bounds: off=%d, len=%d", offset, len(data)) + } + + // Map byte slice directly to the container data. + citer.Next() + _, c := citer.Value() + + // this shouldn't happen, since we don't normally store nils. + if c == nil { + continue + } + switch c.typ() { + case containerRun: + runCount := binary.LittleEndian.Uint16(data[offset : offset+runCountHeaderSize]) + c.setRuns((*[0xFFFFFFF]interval16)(unsafe.Pointer(&data[offset+runCountHeaderSize]))[:runCount:runCount]) + opsOffset = int(offset) + runCountHeaderSize + len(c.runs())*interval16Size + case containerArray: + c.setArray((*[0xFFFFFFF]uint16)(unsafe.Pointer(&data[offset]))[:c.N():c.N()]) + opsOffset = int(offset) + len(c.array())*2 // sizeof(uint32) + case containerBitmap: + c.setBitmap((*[0xFFFFFFF]uint64)(unsafe.Pointer(&data[offset]))[:bitmapN:bitmapN]) + opsOffset = int(offset) + len(c.bitmap())*8 // sizeof(uint64) + } + } + + // Read ops log until the end of the file. + buf := data[opsOffset:] + + for { + // Exit when there are no more ops to parse. + if len(buf) == 0 { + break + } + // Unmarshal the op and apply it. + var opr op + if err := opr.UnmarshalBinary(buf); err != nil { + // FIXME(benbjohnson): return error with position so file can be trimmed. + return err + } + opr.apply(b) + // Increase the op count. + b.ops++ + b.opN += opr.count() + // Move the buffer forward. + buf = buf[opr.size():] + } + + return nil +} diff --git a/server.go b/server.go index 6eee374d1..10de7fc91 100644 --- a/server.go +++ b/server.go @@ -388,6 +388,35 @@ func NewServer(opts ...ServerOption) (*Server, error) { return s, nil } +// UpAndDown brings the server up minimally and shuts it down +// again; basically, it exists for testing holder open and close. +func (s *Server) UpAndDown() error { + s.logger.Printf("open server") + + // Log startup + err := s.holder.logStartup() + if err != nil { + log.Println(errors.Wrap(err, "logging startup")) + } + + // Initialize id-key storage. + if err := s.holder.translateFile.Open(); err != nil { + return errors.Wrap(err, "opening TranslateFile") + } + + // Open holder. + if err := s.holder.Open(); err != nil { + return errors.Wrap(err, "opening Holder") + } + + errh := s.holder.Close() + if errh != nil { + return errors.Wrap(errh, "closing holder") + } + + return nil +} + // Open opens and initializes the server. func (s *Server) Open() error { s.logger.Printf("open server") diff --git a/server/server.go b/server/server.go index b4f05bbba..6add61591 100644 --- a/server/server.go +++ b/server/server.go @@ -161,6 +161,38 @@ func (m *Command) Start() (err error) { return nil } +func (m *Command) UpAndDown() (err error) { + // Seed random number generator + rand.Seed(time.Now().UTC().UnixNano()) + + // SetupServer + err = m.SetupServer() + if err != nil { + return errors.Wrap(err, "setting up server") + } + + // SetupNetworking (so we'll have profiling) + err = m.setupNetworking() + if err != nil { + return errors.Wrap(err, "setting up networking") + } + go func() { + err := m.Handler.Serve() + if err != nil { + m.logger.Printf("handler serve error: %v", err) + } + }() + + // Bring the server up, and back down again. + if err = m.Server.UpAndDown(); err != nil { + return errors.Wrap(err, "bringing server up and down") + } + + m.logger.Printf("brought up and shut down again") + + return nil +} + // Wait waits for the server to be closed or interrupted. func (m *Command) Wait() error { // First SIGKILL causes server to shut down gracefully. diff --git a/view.go b/view.go index 4bde632f9..60841c190 100644 --- a/view.go +++ b/view.go @@ -15,9 +15,11 @@ package pilosa import ( + "context" "fmt" "os" "path/filepath" + "runtime" "strconv" "strings" "sync" @@ -28,6 +30,7 @@ import ( "github.com/pilosa/pilosa/roaring" "github.com/pilosa/pilosa/stats" "github.com/pkg/errors" + "golang.org/x/sync/errgroup" ) // View layout modes. @@ -111,6 +114,8 @@ func (v *view) open() error { return nil } +var workQueue = make(chan struct{}, runtime.NumCPU()*2) + // openFragments opens and initializes the fragments inside the view. func (v *view) openFragments() error { file, err := os.Open(filepath.Join(v.path, "fragments")) @@ -126,29 +131,47 @@ func (v *view) openFragments() error { return errors.Wrap(err, "reading fragments directory") } - for _, fi := range fis { - if fi.IsDir() { - continue - } + eg, ctx := errgroup.WithContext(context.Background()) + var mu sync.Mutex - // Parse filename into integer. - shard, err := strconv.ParseUint(filepath.Base(fi.Name()), 10, 64) - if err != nil { - v.logger.Debugf("WARNING: couldn't use non-integer file as shard in index/field/view %s/%s/%s: %s", v.index, v.field, v.name, fi.Name()) - continue - } + for _, loopFi := range fis { + select { + case <-ctx.Done(): + break + default: + fi := loopFi - v.logger.Debugf("open index/field/view/fragment: %s/%s/%s/%d", v.index, v.field, v.name, shard) - frag := v.newFragment(v.fragmentPath(shard), shard) - if err := frag.Open(); err != nil { - return fmt.Errorf("open fragment: shard=%d, err=%s", frag.shard, err) + if fi.IsDir() { + continue + } + + // Parse filename into integer. + shard, err := strconv.ParseUint(filepath.Base(fi.Name()), 10, 64) + if err != nil { + v.logger.Debugf("WARNING: couldn't use non-integer file as shard in index/field/view %s/%s/%s: %s", v.index, v.field, v.name, fi.Name()) + continue + } + + workQueue <- struct{}{} + v.logger.Debugf("open index/field/view/fragment: %s/%s/%s/%d", v.index, v.field, v.name, shard) + eg.Go(func() error { + defer func() { + <-workQueue + }() + frag := v.newFragment(v.fragmentPath(shard), shard) + if err := frag.Open(); err != nil { + return fmt.Errorf("open fragment: shard=%d, err=%s", frag.shard, err) + } + frag.RowAttrStore = v.rowAttrStore + v.logger.Debugf("add index/field/view/fragment to view.fragments: %s/%s/%s/%d", v.index, v.field, v.name, shard) + mu.Lock() + v.fragments[frag.shard] = frag + mu.Unlock() + return nil + }) } - frag.RowAttrStore = v.rowAttrStore - v.logger.Debugf("add index/field/view/fragment to view.fragments: %s/%s/%s/%d", v.index, v.field, v.name, shard) - v.fragments[frag.shard] = frag } - - return nil + return eg.Wait() } // close closes the view and its fragments. @@ -157,14 +180,29 @@ func (v *view) close() error { defer v.mu.Unlock() // Close all fragments. - for _, frag := range v.fragments { - if err := frag.Close(); err != nil { - return errors.Wrap(err, "closing fragment") + eg, ctx := errgroup.WithContext(context.Background()) + for _, loopFrag := range v.fragments { + select { + case <-ctx.Done(): + break + default: + frag := loopFrag + workQueue <- struct{}{} + eg.Go(func() error { + defer func() { + <-workQueue + }() + + if err := frag.Close(); err != nil { + return errors.Wrap(err, "closing fragment") + } + return nil + }) } } + err := eg.Wait() v.fragments = make(map[uint64]*fragment) - - return nil + return err } // flags returns a set of flags for the underlying fragments.