scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-07 00:55:55 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1988 from seebs/startupspeed

Browse source 
WIP: address some startup speed and performance issues
This commit is contained in:
seebs 2019-08-07 12:38:52 -05:00 • committed by GitHub
commit bc9747cc0f
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
14 changed files with 520 additions and 290 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
cmd/root.go
View file

@ -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

27
cmd/server.go
View file

@ -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 {

5
ctl/check_test.go
View file

@ -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
}

5
ctl/inspect_test.go
View file

@ -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()

77
field.go
View file

@ -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.

2
fragment.go
View file

@ -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...

51
index.go
View file

@ -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.

28
roaring/fuzz_test.go
View file

@ -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)
}
}

239
roaring/roaring.go
View file

@ -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.

6
roaring/roaring_internal_test.go
View file

@ -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)
}
}

222
roaring/unmarshal_binary.go Normal file
View file

@ -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
}

29
server.go
View file

@ -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")

32
server/server.go
View file

@ -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.

86
view.go
View file

@ -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.