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

Use work queue for opening/closing fragments

Browse source 
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.
This commit is contained in:
Seebs 2019-07-23 17:11:32 -05:00
parent b04037900c
commit 2d9ca0888f
3 changed files with 147 additions and 67 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

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.

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.