From 29a1db45504160d6d73f77b43f8c13d17ee83470 Mon Sep 17 00:00:00 2001 From: Seebs Date: Tue, 23 Jul 2019 17:03:55 -0500 Subject: [PATCH 1/6] add "holder" command to start up and shut down It would be neat to be able to observe performance of "just open the holder". So let's make that a verb. --- cmd/root.go | 1 + cmd/server.go | 27 +++++++++++++++++++++++++++ server.go | 29 +++++++++++++++++++++++++++++ server/server.go | 32 ++++++++++++++++++++++++++++++++ 4 files changed, 89 insertions(+) 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/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 db367f418..ecc2f4651 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. From b04037900cc6b493d97f538adf1a86ecaf801cb0 Mon Sep 17 00:00:00 2001 From: Seebs Date: Wed, 24 Jul 2019 14:08:50 -0500 Subject: [PATCH 2/6] move to using roaring iterators for UnmarshalBinary The new roaring iterator used for the remap and importroaring things could also be used for unmarshalling roaring streams, and it's a slightly simpler design that doesn't need two passes through the data. This patch cleans that up a bit, makes it work better with ops logs, and uses that instead. It appears to noticably but not immensely reduce the time imports take, but it also gets us back down to one thing parsing roaring formats. There are a couple of subtle changes to errors we were testing for in various tests, and one of the fuzz tests goes away because it was actually itself an erroneous error message -- it was reporting the header of a roaring file as an invalid op because the op log reader was running on the header for roaring files with zero containers. Oops. --- ctl/check_test.go | 5 +- ctl/inspect_test.go | 5 +- fragment.go | 2 + roaring/fuzz_test.go | 27 ++--- roaring/roaring.go | 189 +++++++++++++------------------ roaring/roaring_internal_test.go | 4 +- 6 files changed, 101 insertions(+), 131 deletions(-) diff --git a/ctl/check_test.go b/ctl/check_test.go index e37d99358..144347685 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: unknown roaring magic number 12849" + 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..d661aa1ab 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: unknown roaring magic number 12849" + if err != nil && err.Error() != expectedError { + t.Fatalf("expected error '%s', got '%v'", expectedError, err) } w.Close() 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/roaring/fuzz_test.go b/roaring/fuzz_test.go index fe8fb6cee..8d934c209 100644 --- a/roaring/fuzz_test.go +++ b/roaring/fuzz_test.go @@ -25,53 +25,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: "reading official 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: want 12935430920 bytes, got 32", }, { 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: want 24696061960 bytes, got 32", }, { 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: want 10200547336 bytes, got 20", }, { 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: want 328 bytes, got 80", }, { // 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: want 56 bytes, got 20", }, { // 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: "container 0/1, expect run length at 9/10 bytes", }, { // 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 (need 12 bytes, found 1)", }, } 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 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..6b5b8149e 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("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,78 +1589,62 @@ func (b *Bitmap) ImportRoaringBits(data []byte, clear bool, log bool, rowSize ui err = b.writeOp(&op) } return changed, rowSet, err +} +func (b *Bitmap) PreferMapping(preferred bool) { + b.preferMapping = preferred } // 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") +func (b *Bitmap) UnmarshalBinary(data []byte) (err error) { + if data == nil { + return errors.New("no roaring bitmap provided") + } + var itr roaringIterator + var itrKey uint64 + var itrCType byte + var itrN int + var itrLen int + var itrPointer *uint16 + var itrErr error + + itr, err = newRoaringIterator(data) + if err != nil { + return err + } + if itr == nil { + return errors.New("failed to create roaring iterator, but don't know why") } - // 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) - } + b.Containers.Reset() - 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)) + itrKey, itrCType, itrN, itrLen, itrPointer, itrErr = itr.Next() + for itrErr == nil { + newC := &Container{ + typeID: itrCType, + n: int32(itrN), + len: int32(itrLen), + cap: int32(itrLen), + pointer: itrPointer, + flags: flagMapped, } - - // 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) + if !b.preferMapping { + newC.unmapOrClone() } + b.Containers.Put(itrKey, newC) + itrKey, itrCType, itrN, itrLen, itrPointer, itrErr = itr.Next() + } + // note: if we get a non-EOF err, it's possible that we made SOME + // changes but didn't log them. I don't have a good solution to this. + if itrErr != io.EOF { + return itrErr } // Read ops log until the end of the file. - buf := data[opsOffset:] + b.ops = 0 + b.opN = 0 + buf := itr.Remaining() for { // Exit when there are no more ops to parse. if len(buf) == 0 { @@ -1648,7 +1667,6 @@ func (b *Bitmap) unmarshalPilosaRoaring(data []byte) error { // Move the buffer forward. buf = buf[opr.size():] } - return nil } @@ -5138,56 +5156,7 @@ 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 diff --git a/roaring/roaring_internal_test.go b/roaring/roaring_internal_test.go index 22b0c3f03..f581f4bff 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: "reading official 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: "reading official header: malformed bitmap, key-cardinality slice overruns buffer at 9", }, { // Runs a bitmap in the Pilosa format through the Pilosa roaring hexString: "3C30000000000000", From 2d9ca0888f09580c2c6b50f2d0de854e151b54ab Mon Sep 17 00:00:00 2001 From: Seebs Date: Tue, 23 Jul 2019 17:11:32 -0500 Subject: [PATCH 3/6] Use work queue for opening/closing fragments When starting up, we can have a large number of views, each with some number of fragments, and by default these were being opened sequentially. There's no real benefit to that; they're all nicely independent from each other and don't need much locking, so we implement a trivial semaphore and launch the operations asynchronously. We also combine them into errgroups. Similarly, we do this for fields and views, capping the number of fields (or views) opened in parallel to avoid hitting a system-wide limit on threads created (oops). Note that the limits are shared, not multiplicative; we cap this fairly arbitrarily at 8 fields being opened, and 16 views being opened, at a time, but NumCPU*2 fragments being opened by those views. This dramatically increases CPU load during startup, but doesn't seem to significantly increase total CPU time, it just scales much better on machines with lots of cores. --- field.go | 77 +++++++++++++++++++++++++++++++------------------- index.go | 51 ++++++++++++++++++++++++--------- view.go | 86 ++++++++++++++++++++++++++++++++++++++++---------------- 3 files changed, 147 insertions(+), 67 deletions(-) diff --git a/field.go b/field.go index 9b49d03b6..b34628398 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. @@ -431,6 +432,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")) @@ -445,40 +448,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/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/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. From 1d732e4b7f72a5e4792be9c0d7bd83d9422fcbf9 Mon Sep 17 00:00:00 2001 From: Seebs Date: Thu, 25 Jul 2019 16:11:57 -0500 Subject: [PATCH 4/6] drop unused functions from previous unmarshal implementation --- roaring/roaring.go | 57 ---------------------------------------------- 1 file changed, 57 deletions(-) diff --git a/roaring/roaring.go b/roaring/roaring.go index 6b5b8149e..56c60d21a 100644 --- a/roaring/roaring.go +++ b/roaring/roaring.go @@ -5156,63 +5156,6 @@ func readOfficialHeader(buf []byte) (size uint32, containerTyper func(index uint return size, containerTyper, header, pos, haveRuns, err } -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. From fba496bc91969b33b9e79f5eb2463fc2515c8ef2 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Mon, 5 Aug 2019 17:39:47 -0500 Subject: [PATCH 5/6] generalize test strings and break out old UnmarshalBinary code (don't use iterator for unmarshalBinary) --- ctl/check_test.go | 2 +- ctl/inspect_test.go | 4 +- roaring/fuzz_test.go | 19 +-- roaring/roaring.go | 77 +--------- roaring/roaring_internal_test.go | 6 +- roaring/roaring_unmarshal_binary.go | 208 ++++++++++++++++++++++++++++ 6 files changed, 225 insertions(+), 91 deletions(-) create mode 100644 roaring/roaring_unmarshal_binary.go diff --git a/ctl/check_test.go b/ctl/check_test.go index 144347685..ab5ef9279 100644 --- a/ctl/check_test.go +++ b/ctl/check_test.go @@ -93,7 +93,7 @@ func TestCheckCommand_Run(t *testing.T) { t.Fatalf("copy: %v", err) } - expectedPrefix := "checking bitmap: unmarshalling: unknown roaring magic number 12849" + expectedPrefix := "checking bitmap: unmarshalling: " if !strings.HasPrefix(err.Error(), expectedPrefix) { t.Fatalf("expect error: '%s...', actual: '%s'", expectedPrefix, err) } diff --git a/ctl/inspect_test.go b/ctl/inspect_test.go index d661aa1ab..1ff852712 100644 --- a/ctl/inspect_test.go +++ b/ctl/inspect_test.go @@ -41,8 +41,8 @@ func TestInspectCommand_Run(t *testing.T) { file.Close() cm.Path = file.Name() err = cm.Run(context.Background()) - expectedError := "unmarshalling: unknown roaring magic number 12849" - if err != nil && err.Error() != expectedError { + expectedError := "unmarshalling: " + if !strings.Contains(err.Error(), expectedError) { t.Fatalf("expected error '%s', got '%v'", expectedError, err) } diff --git a/roaring/fuzz_test.go b/roaring/fuzz_test.go index 8d934c209..86f99faa0 100644 --- a/roaring/fuzz_test.go +++ b/roaring/fuzz_test.go @@ -14,6 +14,7 @@ package roaring import ( + "strings" "testing" ) @@ -25,41 +26,41 @@ func TestUnmarshalBinary(t *testing.T) { }{ { // Checks for the zero containers situation cr: []byte(":0\x00\x00\x01\x00\x00\x000000"), //":000000" - expected: "reading official header: malformed bitmap, key-cardinality slice overruns buffer at 12", + 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: "insufficient data for header + offsets: want 12935430920 bytes, got 32", + 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: "insufficient data for header + offsets: want 24696061960 bytes, got 32", + expected: "insufficient data for header + offsets:", }, { cr: []byte("<0\x00\x02\x00\x00\x00&x.field safe"), //"<0&x.field safe" - expected: "insufficient data for header + offsets: want 10200547336 bytes, got 20", + 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: "insufficient data for header + offsets: want 328 bytes, got 80", + 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: "insufficient data for header + offsets: want 56 bytes, got 20", + expected: "insufficient data for header + offsets:", }, { // Checks for incomplete offset in readWithRuns cr: []byte(";0\x00\x00\v00000"), //";00 00000" - expected: "container 0/1, expect run length at 9/10 bytes", + expected: "insufficient data for offsets", }, { // Checks for incomplete offset in readOffsets cr: []byte(":0\x00\x00\x03\x00\x00\x00000000000000" + "\x00"), //:0000000000000 - expected: "insufficient data for offsets (need 12 bytes, found 1)", + expected: "insufficient data for offsets", }, } @@ -67,7 +68,7 @@ func TestUnmarshalBinary(t *testing.T) { err := b.UnmarshalBinary(crash.cr) if err == nil { t.Errorf("expected: %s, got: no error", crash.expected) - } else if err.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 56c60d21a..86ee64fc5 100644 --- a/roaring/roaring.go +++ b/roaring/roaring.go @@ -1364,7 +1364,7 @@ func (r *officialRoaringIterator) Next() (key uint64, cType byte, n int, length var runCount uint16 if r.currentType == containerRun { if int(r.currentDataOffset)+2 > len(r.data) { - r.Done(fmt.Errorf("container %d/%d, expect run length at %d/%d bytes", + 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() } @@ -1595,81 +1595,6 @@ func (b *Bitmap) PreferMapping(preferred bool) { b.preferMapping = preferred } -// unmarshalPilosaRoaring treats data as being encoded in Pilosa's 64 bit -// roaring format and decodes it into b. -func (b *Bitmap) UnmarshalBinary(data []byte) (err error) { - if data == nil { - return errors.New("no roaring bitmap provided") - } - var itr roaringIterator - var itrKey uint64 - var itrCType byte - var itrN int - var itrLen int - var itrPointer *uint16 - var itrErr error - - itr, err = newRoaringIterator(data) - if err != nil { - return err - } - if itr == nil { - return errors.New("failed to create roaring iterator, but don't know why") - } - - b.Containers.Reset() - - itrKey, itrCType, itrN, itrLen, itrPointer, itrErr = itr.Next() - for itrErr == nil { - newC := &Container{ - typeID: itrCType, - n: int32(itrN), - len: int32(itrLen), - cap: int32(itrLen), - pointer: itrPointer, - flags: flagMapped, - } - if !b.preferMapping { - newC.unmapOrClone() - } - b.Containers.Put(itrKey, newC) - itrKey, itrCType, itrN, itrLen, itrPointer, itrErr = itr.Next() - } - // note: if we get a non-EOF err, it's possible that we made SOME - // changes but didn't log them. I don't have a good solution to this. - if itrErr != io.EOF { - return itrErr - } - - // Read ops log until the end of the file. - b.ops = 0 - b.opN = 0 - buf := itr.Remaining() - 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 -} - // writeOp writes op to the OpWriter, if available. func (b *Bitmap) writeOp(op *op) error { if b.OpWriter == nil { diff --git a/roaring/roaring_internal_test.go b/roaring/roaring_internal_test.go index f581f4bff..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 official 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 official 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/roaring_unmarshal_binary.go b/roaring/roaring_unmarshal_binary.go new file mode 100644 index 000000000..e7e0fc101 --- /dev/null +++ b/roaring/roaring_unmarshal_binary.go @@ -0,0 +1,208 @@ +// +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 +} From b84eada5215879a5aa34f2af5958e1db840b43df Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Wed, 7 Aug 2019 08:09:04 -0500 Subject: [PATCH 6/6] rename file and add license header --- ...ing_unmarshal_binary.go => unmarshal_binary.go} | 14 ++++++++++++++ 1 file changed, 14 insertions(+) rename roaring/{roaring_unmarshal_binary.go => unmarshal_binary.go} (92%) diff --git a/roaring/roaring_unmarshal_binary.go b/roaring/unmarshal_binary.go similarity index 92% rename from roaring/roaring_unmarshal_binary.go rename to roaring/unmarshal_binary.go index e7e0fc101..6b13cce0f 100644 --- a/roaring/roaring_unmarshal_binary.go +++ b/roaring/unmarshal_binary.go @@ -1,3 +1,17 @@ +// 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