diff --git a/holder.go b/holder.go index e02734be2..f93be5ba7 100644 --- a/holder.go +++ b/holder.go @@ -666,6 +666,7 @@ func (h *Holder) Open() error { if err := h.txf.green2blue(h); err != nil { return errors.Wrap(err, "Holder.Open h.txf.green2blue(h)") } + h.txf.blueGreenOnIfRunningBlueGreen() h.Logger.Printf("open holder: complete") diff --git a/index.go b/index.go index 44200b191..260054f59 100644 --- a/index.go +++ b/index.go @@ -26,7 +26,6 @@ import ( "sync" "time" - "github.com/glycerine/idem" "github.com/gogo/protobuf/proto" "github.com/pilosa/pilosa/v2/hash" "github.com/pilosa/pilosa/v2/internal" @@ -799,55 +798,13 @@ func (idx *Index) ComputeTranslatorSummary(verbose, checkKeys, applyKeyRepairs b fmt.Printf("\n# index: %v\n# =================\n", idx.name) } - jobQ := make(chan func() error, 10000) - var errmu sync.Mutex - - 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) - } + pjob := newParallelJobs(parallelReaders) floop: for _, fld := range idx.fields { fld := fld - fun := func() error { + fun := func(worker int) error { //vv("ComputeTranslatorSummary() on fld '%v'", fld.name) sum, err := fld.translateStore.ComputeTranslatorSummaryRows() if err != nil { @@ -866,10 +823,8 @@ floop: return nil } - select { - case <-halters[0].ReqStop.Chan: + if !pjob.run(fun) { break floop - case jobQ <- fun: } } // end floop @@ -882,7 +837,7 @@ tloop: partitionID := partitionID store := store - fun2 := func() error { + fun2 := func(worker int) error { //vv("ComputeTranslatorSummary() running on store.Path = '%v'", store.GetStorePath()) if checkKeys { prim := topo.PrimaryNodeIndex(partitionID) @@ -943,19 +898,14 @@ tloop: return nil } - select { - case <-halters[0].ReqStop.Chan: + if !pjob.run(fun2) { break tloop - case jobQ <- fun2: } + } // end tloop - close(jobQ) // tell the workers no more jobs. + err = pjob.waitForFinish() - // wait for everyone to finish - for _, h := range halters { - <-h.Done.Chan - } return ats, err } diff --git a/pjobs.go b/pjobs.go new file mode 100644 index 000000000..0636fbd36 --- /dev/null +++ b/pjobs.go @@ -0,0 +1,115 @@ +// 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. + +package pilosa + +import ( + "sync" + + "github.com/glycerine/idem" +) + +// parallelJobs runs functions in parallel on a goroutine +// pool that has nGoro goroutines. +type parallelJobs struct { + nGoro int + + jobQ chan func(worker int) error + halters []*idem.Halter + + // err is protected by errmu + err error + errmu sync.Mutex +} + +func newParallelJobs(nGoro int) (p *parallelJobs) { + if nGoro < 1 { + // 0 really means, + // "turn it up to 11". + // same for negative. + nGoro = 10000 + } + // maximum 10K goroutines + if nGoro > 10000 { + nGoro = 10000 + } + + p = ¶llelJobs{ + nGoro: nGoro, + jobQ: make(chan func(worker int) error, 10000), + halters: make([]*idem.Halter, nGoro), + } + + for j := 0; j < nGoro; j++ { + h := idem.NewHalter() + p.halters[j] = h + } + + for i, h := range p.halters { + go func(h *idem.Halter, worker int) { + defer h.MarkDone() + for { + select { + case <-h.ReqStop.Chan: + return + case f, ok := <-p.jobQ: + if !ok { + // channel closed, finish up + return + } + + err1 := f(worker) + if err1 != nil { + p.errmu.Lock() + if p.err == nil { + p.err = err1 + } + p.errmu.Unlock() + // an error occurred, tell everyone to stop + for _, h2 := range p.halters { + h2.RequestStop() + } + return + } + } + } + }(h, i) + } + return +} + +// return value accepted will be false if we are shutting down +// due to an error. +func (p *parallelJobs) run(fun func(worker int) error) (accepted bool) { + select { + case <-p.halters[0].ReqStop.Chan: + return false + case p.jobQ <- fun: + return true + } +} + +func (p *parallelJobs) waitForFinish() error { + + // tell the workers no more jobs. + close(p.jobQ) + + // wait for everyone to finish + for i, h := range p.halters { + _ = i + <-h.Done.Chan + } + + return p.err +} diff --git a/pjobs_test.go b/pjobs_test.go new file mode 100644 index 000000000..71192b278 --- /dev/null +++ b/pjobs_test.go @@ -0,0 +1,51 @@ +// 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. + +package pilosa + +import ( + "fmt" + "sync/atomic" + "testing" +) + +func Test_ParallelJobs_EarlyShutdown_WaitsForAllGoro(t *testing.T) { + const n = 10000 // total jobs to run + + var errLastOne = fmt.Errorf("the last job has run, and returned this error") + pj := newParallelJobs(100) + nTotal := int64(0) + for i := 0; i < n; i++ { + accepted := pj.run(func(worker int) error { + highpoint := atomic.AddInt64(&nTotal, 1) + switch int(highpoint) { + case n - 1: + return errLastOne + } + return nil + }) + if !accepted { + panic("should have been accepted") + } + } + err := pj.waitForFinish() + tot := atomic.LoadInt64(&nTotal) + if int(tot) != n { + panic(fmt.Sprintf("We didn't run them all? tot=%v, n=%v; pj.jobQ len %v; err='%v'", tot, n, len(pj.jobQ), err)) + } + if err != errLastOne { + panic("expected to see errLastOne") + } + // good: finished cleanly. +} diff --git a/txfactory.go b/txfactory.go index 777fa03ae..70375d61f 100644 --- a/txfactory.go +++ b/txfactory.go @@ -20,12 +20,12 @@ import ( "os" "path" "path/filepath" + "runtime" "strconv" "strings" "sync" "syscall" "text/tabwriter" - "time" "github.com/pilosa/pilosa/v2/hash" "github.com/pilosa/pilosa/v2/rbf" @@ -1320,7 +1320,7 @@ func (f *TxFactory) greenHasData() (hasData bool, err error) { // txfactory_internal_test.go as well. // // This is a noop if we aren't running under a blue_green PILOSA_TXSRC. -func (f *TxFactory) green2blue(holder *Holder) (err error) { +func (f *TxFactory) green2blue(holder *Holder) (err0 error) { // Holder.Open will always call us, even without blue_green. Which is fine. // We are just a no-op in that case. @@ -1360,13 +1360,25 @@ func (f *TxFactory) green2blue(holder *Holder) (err error) { return fmt.Errorf("error: cannot migrate from green '%v' because it has no data in it.", greenSrc) } + nGoro := runtime.NumCPU() + if nGoro < 5 { + // try to get some overlapped IO + nGoro = 5 + } + pj := newParallelJobs(nGoro) + + action := "verify" if blueHasData { verifyInsteadOfCopy = true + defer holder.Logger.Printf("bitmap-backend verification done : %v compared to %v", blueDest, greenSrc) } else { - holder.Logger.Printf("bitmap-backend migration starting: populating %v from %v", blueDest, greenSrc) + action = "migrate" + holder.Logger.Printf("bitmap-backend migration starting: populating %v from %v with %v threads", blueDest, greenSrc, nGoro) defer holder.Logger.Printf("bitmap-backend migration done : populated %v from %v", blueDest, greenSrc) } + firstPjobStarted := false +indexloop: for k, idx := range idxs { // scan directories @@ -1399,39 +1411,56 @@ func (f *TxFactory) green2blue(holder *Holder) (err error) { } } - lastProgress := time.Now() - progressCount := 0 + shardNum := 0 for shard := range greenShards { + shardNum++ + shnum := shardNum + idx := idx + shard := shard + k := k + fun := func(worker int) error { - dbs, err := f.dbPerShard.GetDBShard(idx.name, shard, idx) - if err != nil { - return errors.Wrap(err, fmt.Sprintf("GetDBShard(index='%v', shard='%v')", idx.name, int(shard))) + dbs, err := f.dbPerShard.GetDBShard(idx.name, shard, idx) + if err != nil { + return errors.Wrap(err, fmt.Sprintf("GetDBShard(index='%v', shard='%v')", idx.name, int(shard))) + } + + holder.Logger.Printf("%v progress on index '%v' (%v of %v): on shard '%v' (%v of %v) [worker %v]", + action, idx.name, k+1, len(idxs), shard, shnum, len(greenShards), worker) + + if verifyInsteadOfCopy { + // verify all containers + err = dbs.verifyBlueEqualsGreen() + if err != nil { + return errors.Wrap(err, + fmt.Sprintf("dbs.verifyBlueEqualsGreen(blue='%v', "+ + "green='%v') for index='%v', shard='%v'", + blueDest, greenSrc, idx.name, int(shard))) + } + } else { + // the main copy work + err = dbs.populateBlueFromGreen() + if err != nil { + return errors.Wrap(err, + fmt.Sprintf("dbs.copyGreenToBlue(blue='%v', "+ + "green='%v') for index='%v', shard='%v'", + blueDest, greenSrc, idx.name, int(shard))) + } + } + return nil + } // end of fun definition + + if !pj.run(fun) { + break indexloop } - - if verifyInsteadOfCopy { - // verify all containers - err = dbs.verifyBlueEqualsGreen() - if err != nil { - return errors.Wrap(err, - fmt.Sprintf("dbs.verifyBlueEqualsGreen(blue='%v', "+ - "green='%v') for index='%v', shard='%v'", - blueDest, greenSrc, idx.name, int(shard))) - } - } else { - // the main copy work - progressCount++ - if progressCount == 1 || time.Since(lastProgress) > time.Second { - holder.Logger.Printf("migration progress on index '%v' (%v of %v): on shard %v of %v", - idx.name, k+1, len(idxs), progressCount, len(greenShards)) - lastProgress = time.Now() - } - err = dbs.populateBlueFromGreen() - if err != nil { - return errors.Wrap(err, - fmt.Sprintf("dbs.copyGreenToBlue(blue='%v', "+ - "green='%v') for index='%v', shard='%v'", - blueDest, greenSrc, idx.name, int(shard))) - } + if !firstPjobStarted { + firstPjobStarted = true + defer func() { + err1 := pj.waitForFinish() + if err0 == nil { + err0 = err1 + } + }() } } } diff --git a/txfactory_internal_test.go b/txfactory_internal_test.go index 6fa9c3b9e..580e2956d 100644 --- a/txfactory_internal_test.go +++ b/txfactory_internal_test.go @@ -22,6 +22,7 @@ import ( "time" "github.com/glycerine/lmdb-go/lmdb" + //"github.com/pilosa/pilosa/v2/logger" ) func Test_TxFactory_Qcx_query_context(t *testing.T) { @@ -390,14 +391,13 @@ func Test_TxFactory_verifyBlueEqualsGreen(t *testing.T) { // The Holder.Open should verify blue against green and notice the extra bit. h6 := NewHolder(path, nil) err = h6.Open() - - //vv("h6.Open() had err = '%v', PILOSA_TXSRC='%v'", err, os.Getenv("PILOSA_TXSRC")) //h6.DumpAllShards() if err == nil { h6.Close() t.Fatalf("should have had blue-green verification fail on Holder.Open") } + h6.Close() } }