Merge pull request #327 from CovenantSQL/feature/miner_startup

Parallelize chain synchronization in startup

sqlchain/chain.go

@@ -792,12 +792,11 @@ func (c *Chain) processBlocks(ctx context.Context) {
 }
 
 // Start starts the main process of the sql-chain.
-func (c *Chain) Start() (err error) {
+func (c *Chain) Start() {
 	c.rt.goFunc(c.processBlocks)
 	c.sync()
 	c.rt.goFunc(c.mainCycle)
 	c.rt.startService(c)
-	return
 }
 
 // Stop stops the main process of the sql-chain.

sqlchain/chain_test.go

@@ -275,10 +275,7 @@ func TestMultiChain(t *testing.T) {
 
 	// Start all chain instances
 	for _, v := range chains {
-		if err = v.chain.Start(); err != nil {
-			t.Fatalf("error occurred: %v", err)
-		}
-
+		v.chain.Start()
 		defer func(c *Chain) {
 			// Stop chain main process before exit
 			_ = c.Stop()

worker/db.go

@@ -175,9 +175,8 @@ func NewDatabase(cfg *DBConfig, peers *proto.Peers,
 	}
 	if db.chain, err = sqlchain.NewChain(chainCfg); err != nil {
 		return
-	} else if err = db.chain.Start(); err != nil {
-		return
 	}
+	db.chain.Start()
 
 	// init kayak config
 	kayakWalPath := filepath.Join(cfg.DataDir, KayakWalFileName)

worker/dbms.go

@@ -367,16 +367,30 @@ func (dbms *DBMS) initDatabases(
 	meta *DBMSMeta, profiles map[proto.DatabaseID]*types.SQLChainProfile) (err error,
 ) {
 	currentInstance := make(map[proto.DatabaseID]bool)
+	wg := &sync.WaitGroup{}
+	errCh := make(chan error, len(profiles))
 
 	for id, profile := range profiles {
 		currentInstance[id] = true
 		var instance *types.ServiceInstance
 		if instance, err = dbms.buildSQLChainServiceInstance(profile); err != nil {
 			return
 		}
-		if err = dbms.Create(instance, false); err != nil {
-			return
-		}
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			if err := dbms.Create(instance, false); err != nil {
+				log.WithFields(log.Fields{
+					"id": instance.DatabaseID,
+				}).WithError(err).Error("failed to create database instance")
+				errCh <- errors.Wrapf(err, "failed to create database %s", instance.DatabaseID)
+			}
+		}()
+	}
+	wg.Wait()
+	close(errCh)
+	for err := range errCh {
+		return err // omit any other error after this instance
 	}
 
 	// calculate to drop databases