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.