scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 10:54:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #977 from molecula/readers

Browse source 
pilosa-fsck: the -readers flag controls parallelism
This commit is contained in:
jaten-molecula 2020-10-13 19:14:09 -05:00 • committed by GitHub
commit a997e713dc
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 109 additions and 23 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
boltdb/translate.go
View file

@ -116,6 +116,15 @@ func NewTranslateStore(index, field string, partitionID, partitionN int) *Transl
// Open opens the translate file.
func (s *TranslateStore) Open() (err error) {
// add the path to the problem database if we panic handling it.
defer func() {
r := recover()
if r != nil {
panic(fmt.Sprintf("pilosa/boltdb/TranslateStore.Open(s.Path='%v') panic with '%v'", s.Path, r))
}
}()
if err := os.MkdirAll(filepath.Dir(s.Path), 0777); err != nil {
return errors.Wrapf(err, "mkdir %s", filepath.Dir(s.Path))
} else if s.db, err = bolt.Open(s.Path, 0666, &bolt.Options{Timeout: 1 * time.Second}); err != nil {

2
cmd/pilosa-chk/chk.go
View file

@ -79,7 +79,7 @@ func main() {
const checkKeys = false
const applyKeyRepairs = false
for _, idx := range holder.Indexes() {
asum, err := idx.ComputeTranslatorSummary(verbose, checkKeys, applyKeyRepairs, nil, "fake-nodeID")
asum, err := idx.ComputeTranslatorSummary(verbose, checkKeys, applyKeyRepairs, nil, "fake-nodeID", 10)
if err != nil {
log.Fatal(err)
}

12
cmd/pilosa-fsck/fsck.go
View file

@ -73,6 +73,8 @@ type FsckConfig struct {
ReplicaN int // -replicas
PilosaConfigPath string // -config
ParallelReaders int // -readers
topo *pilosa.Topology
}
@ -85,6 +87,8 @@ func (cfg *FsckConfig) DefineFlags(fs *flag.FlagSet) {
fs.IntVar(&cfg.ReplicaN, "replicas", 0, "(required) manually entered replicaN; the number of replicas maintained in the cluster. Must be the same as the [cluster] 'replicas = R' entry in the pilosa.conf file for the cluster.")
fs.IntVar(&cfg.ParallelReaders, "readers", 10, "how many parallel readers to use to scan at once. 0 means do everything possible in parallel. 1 means serialize everything through a single reader. Can be adjusted to control memory consumption.")
fs.StringVar(&cfg.PilosaConfigPath, "config", "", "(required: -replicas or -config, with -config preferred) path to the pilosa.conf for the cluster (e.g. /etc/pilosa.conf)")
fs.StringVar(&cfg.JustThisIndex, "index", "", "(optional) restrict to just this index. Otherwise we default to all indexes.")
@ -104,6 +108,12 @@ func (cfg *FsckConfig) DefineFlags(fs *flag.FlagSet) {
-index index_name
(optional) restrict to just this index. Otherwise we default to all indexes.
-readers PR
how many parallel readers to use to scan at once. PR==0 means do everything
possible in parallel. PR==1 means serialize everything through a single reader.
Adjust PR to control memory consumption if needed. As a practical limit, setting
PR > 10000 will have no effect. (default is 10).
-q
be very quiet during analysis and repair
@ -633,7 +643,7 @@ func (cfg *FsckConfig) readOneDir(dir string) (idx2frag map[string]*pilosa.Index
//vv("calling idx.ComputeTranslatorSummary(verbose, checkKeys=%v, cfg.FixCol='%v')", checkKeys, cfg.FixCol)
asum, err := idx.ComputeTranslatorSummary(verbose, checkKeys, cfg.FixCol, topo, nodeID)
asum, err := idx.ComputeTranslatorSummary(verbose, checkKeys, cfg.FixCol, topo, nodeID, cfg.ParallelReaders)
if err != nil {
log.Fatal(err)
}

5
cmd/pilosa-fsck/fsck_test.go
View file

@ -201,8 +201,9 @@ func Test_Repair(t *testing.T) {
FixCol: false,
Quiet: true,
//Verbose: true,
ReplicaN: nReplicas,
Dirs: dirs,
ReplicaN: nReplicas,
Dirs: dirs,
ParallelReaders: 5,
}
panicOn(cfg.ValidateConfig())

1
go.mod
View file

@ -13,6 +13,7 @@ require (
github.com/davecgh/go-spew v1.1.1
github.com/dgraph-io/badger/v2 v2.0.1-rc1.0.20200709123515-8e896a7af361
github.com/dustin/go-humanize v1.0.0
github.com/glycerine/idem v0.0.0-20190127113923-7a8083893311
github.com/glycerine/lmdb-go v1.9.32
github.com/go-ole/go-ole v1.2.4 // indirect
github.com/gogo/protobuf v1.2.1

103
index.go
View file

@ -26,6 +26,7 @@ import (
"sync"
"time"
"github.com/glycerine/idem"
"github.com/gogo/protobuf/proto"
"github.com/pilosa/pilosa/v2/hash"
"github.com/pilosa/pilosa/v2/internal"
@ -787,30 +788,74 @@ func (ats *AllTranslatorSummary) Sort() {
}
// sums is only guaranteed to be sorted by (index, PartitionID, field) iff err returns nil
func (i *Index) ComputeTranslatorSummary(verbose, checkKeys, applyKeyRepairs bool, topo *Topology, nodeID string) (ats *AllTranslatorSummary, err error) {
i.mu.RLock()
defer i.mu.RUnlock()
func (idx *Index) ComputeTranslatorSummary(verbose, checkKeys, applyKeyRepairs bool, topo *Topology, nodeID string, parallelReaders int) (ats *AllTranslatorSummary, err error) {
idx.mu.RLock()
defer idx.mu.RUnlock()
ats = &AllTranslatorSummary{}
var atsMu sync.Mutex
if verbose {
fmt.Printf("\n# index: %v\n# =================\n", i.name)
fmt.Printf("\n# index: %v\n# =================\n", idx.name)
}
var g errgroup.Group
jobQ := make(chan func() error, 10000)
var errmu sync.Mutex
for _, fld := range i.fields {
if parallelReaders < 1 {
// turn it up to 11
parallelReaders = 10000
}
halters := make([]*idem.Halter, parallelReaders)
for j := 0; j < parallelReaders; j++ {
h := idem.NewHalter()
halters[j] = h
}
for _, h := range halters {
go func(h *idem.Halter) {
defer h.MarkDone()
for {
select {
case <-h.ReqStop.Chan:
return
case f, ok := <-jobQ:
if !ok || f == nil {
// channel closed, finish up
return
}
err1 := f()
if err1 != nil {
errmu.Lock()
if err == nil {
err = err1
}
errmu.Unlock()
// an error occurred, tell everyone to stop
for _, h2 := range halters {
h2.RequestStop()
}
return
}
}
}
}(h)
}
floop:
for _, fld := range idx.fields {
fld := fld
g.Go(func() error {
//vv("g.Go() on fld '%v'", fld.name)
fun := func() error {
//vv("ComputeTranslatorSummary() on fld '%v'", fld.name)
sum, err := fld.translateStore.ComputeTranslatorSummaryRows()
if err != nil {
return err
}
sum.Field = fld.name
sum.Index = i.Name()
sum.Checksum = hash.Blake3sum16([]byte(fmt.Sprintf("%v/%v/%v", sum.Checksum, fld.name, i.Name())))
sum.Index = idx.Name()
sum.Checksum = hash.Blake3sum16([]byte(fmt.Sprintf("%v/%v/%v", sum.Checksum, fld.name, idx.Name())))
sum.IsColKey = false
if verbose {
fmt.Printf("# row blake3-%v keyN: %5v idN: %5v field: '%v'\n", sum.Checksum, sum.KeyCount, sum.IDCount, fld.name)
@ -819,17 +864,26 @@ func (i *Index) ComputeTranslatorSummary(verbose, checkKeys, applyKeyRepairs boo
ats.Sums = append(ats.Sums, sum)
atsMu.Unlock()
return nil
})
}
}
select {
case <-halters[0].ReqStop.Chan:
break floop
case jobQ <- fun:
}
} // end floop
if verbose {
fmt.Printf("# ====================\n")
}
for partitionID, store := range i.translateStores {
tloop:
for partitionID, store := range idx.translateStores {
partitionID := partitionID
store := store
g.Go(func() error {
//vv("g.Go() running on store.Path = '%v'", store.GetStorePath())
fun2 := func() error {
//vv("ComputeTranslatorSummary() running on store.Path = '%v'", store.GetStorePath())
if checkKeys {
prim := topo.PrimaryNodeIndex(partitionID)
primID := topo.nodeIDs[prim]
@ -864,7 +918,7 @@ func (i *Index) ComputeTranslatorSummary(verbose, checkKeys, applyKeyRepairs boo
}
sum.IsColKey = true
sum.PartitionID = partitionID
sum.Index = i.Name()
sum.Index = idx.Name()
sum.StorePath = store.GetStorePath()
sum.NodeID = nodeID
sum.IsPrimary = topo.IsPrimary(nodeID, partitionID)
@ -877,7 +931,7 @@ func (i *Index) ComputeTranslatorSummary(verbose, checkKeys, applyKeyRepairs boo
}
}
sum.Checksum = hash.Blake3sum16([]byte(fmt.Sprintf("%v/%v/%v", sum.Checksum, partitionID, i.Name())))
sum.Checksum = hash.Blake3sum16([]byte(fmt.Sprintf("%v/%v/%v", sum.Checksum, partitionID, idx.Name())))
if verbose {
// This is not regular index logging. This is output of the pilosa-fsck tool.
// So it must be printing straight to stdout.
@ -888,9 +942,20 @@ func (i *Index) ComputeTranslatorSummary(verbose, checkKeys, applyKeyRepairs boo
atsMu.Unlock()
return nil
})
}
select {
case <-halters[0].ReqStop.Chan:
break tloop
case jobQ <- fun2:
}
} // end tloop
close(jobQ) // tell the workers no more jobs.
// wait for everyone to finish
for _, h := range halters {
<-h.Done.Chan
}
err = g.Wait()
return ats, err
}