Merge pull request #1071 from jaten-molecula/parallel_migration_rb

run migration in parallel

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

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
 }
 

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 = &parallelJobs{
+		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
+}

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.
+}

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
+					}
+				}()
 			}
 		}
 	}

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