Merge pull request #339 from meshplus/fix-retry-start-plugin

Fix retry start plugin

Author: jiuhuche120

go.sum

@@ -697,8 +697,6 @@ github.com/mattn/go-runewidth v0.0.9/go.mod h1:H031xJmbD/WCDINGzjvQ9THkh0rPKHF+m
 github.com/mattn/go-sqlite3 v1.11.0/go.mod h1:FPy6KqzDD04eiIsT53CuJW3U88zkxoIYsOqkbpncsNc=
 github.com/mattn/go-tty v0.0.0-20180907095812-13ff1204f104/go.mod h1:XPvLUNfbS4fJH25nqRHfWLMa1ONC8Amw+mIA639KxkE=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
-github.com/meshplus/bitxhub-core v1.3.1-0.20220419015433-0fc9632c8a9c h1:LNX5HHIw2iHdT2+Hxvfo7TWU3fAgL46iEHNVRq7Xdro=
-github.com/meshplus/bitxhub-core v1.3.1-0.20220419015433-0fc9632c8a9c/go.mod h1:4XwpJi5KRDb7Uwzf8mCy3XZVrLw2HOA3pngpvyGLtLU=
 github.com/meshplus/bitxhub-core v1.3.1-0.20220419072638-73c914bd56f1 h1:R0cssmTlWe8yeRiEjw/zo/q8HLh3To+mX5mqrfgpDJM=
 github.com/meshplus/bitxhub-core v1.3.1-0.20220419072638-73c914bd56f1/go.mod h1:Rkokix/QLfDhQBTtr5jvFNJlxnPH4tb3oZhgTXDP3iY=
 github.com/meshplus/bitxhub-kit v1.0.0-rc1/go.mod h1:ra/AhOkPvpElI+wXrB9G6DjdcrdxFU3vMwA5MYKr9D0=
@@ -708,10 +706,8 @@ github.com/meshplus/bitxhub-kit v1.2.1-0.20220325052414-bc17176c509d/go.mod h1:w
 github.com/meshplus/bitxhub-kit v1.2.1-0.20220412092457-5836414df781 h1:xFu/v4zA+nbveKVc/GxzCZr8opP7AzDL79mWivAY/DA=
 github.com/meshplus/bitxhub-kit v1.2.1-0.20220412092457-5836414df781/go.mod h1:wrEdhHp1tktzdwcWb4bOxYsVc+KkcrYL18IYWYeumPQ=
 github.com/meshplus/bitxhub-model v1.0.0-rc3/go.mod h1:ZCctQIYTlE3vJ8Lhkrgs9bWwNA+Dw4JzojOSIzLVU6E=
-github.com/meshplus/bitxhub-model v1.2.1-0.20220407083019-e55040281b3f/go.mod h1:sPko8fD+G3MYOnH/44Ju4T4jD/tlEU3IJDIJ4hjrzxg=
 github.com/meshplus/bitxhub-model v1.2.1-0.20220412064024-c35cae241eb2/go.mod h1:sPko8fD+G3MYOnH/44Ju4T4jD/tlEU3IJDIJ4hjrzxg=
-github.com/meshplus/bitxhub-model v1.2.1-0.20220415095224-3a9571aa1d15 h1:8BP+MSBMq2r5UquhNrAA/ZOdNLmCgNWBUnvZpxIWAjU=
-github.com/meshplus/bitxhub-model v1.2.1-0.20220415095224-3a9571aa1d15/go.mod h1:sPko8fD+G3MYOnH/44Ju4T4jD/tlEU3IJDIJ4hjrzxg=
+github.com/meshplus/bitxhub-model v1.2.1-0.20220419071920-bd5cc4a17768 h1:OMfyCVQAHsGYh/J24MQmJfrcMxCKwZ9nmWlqBMyS+s8=
 github.com/meshplus/bitxhub-model v1.2.1-0.20220419071920-bd5cc4a17768/go.mod h1:sPko8fD+G3MYOnH/44Ju4T4jD/tlEU3IJDIJ4hjrzxg=
 github.com/meshplus/go-bitxhub-client v1.4.1-0.20220412093230-11ca79f069fc h1:3J52XPtUFf0gvJHn61EBr+pI268/edQv544w+TWgEwU=
 github.com/meshplus/go-bitxhub-client v1.4.1-0.20220412093230-11ca79f069fc/go.mod h1:pvV1elDWb8ITsOccfJa6CXEHHHb7yURmtGSOngTPL8c=

internal/adapt/appchain_adapter/adapter.go

@@ -3,10 +3,7 @@ package appchain_adapter
 import (
 	"fmt"
 	"strings"
-	"time"
 
-	"github.com/Rican7/retry"
-	"github.com/Rican7/retry/strategy"
 	"github.com/hashicorp/go-plugin"
 	"github.com/meshplus/bitxhub-model/pb"
 	"github.com/meshplus/pier/internal/adapt"
@@ -312,15 +309,15 @@ func findRemoteInterchain(remoteServiceID string, outMeta map[string]uint64, cal
 func (a *AppchainAdapter) init() error {
 	var err error
 
-	if err := retry.Retry(func(attempt uint) error {
-		a.client, a.pluginClient, err = plugins.CreateClient(&a.config.Appchain, nil)
-		if err != nil {
-			a.logger.Errorf("create client plugin", "error", err.Error())
-		}
-		return err
-	}, strategy.Wait(3*time.Second)); err != nil {
-		return fmt.Errorf("retry error to create plugin: %w", err)
+	//if err := retry.Retry(func(attempt uint) error {
+	a.client, a.pluginClient, err = plugins.CreateClient(&a.config.Appchain, nil)
+	if err != nil {
+		a.logger.Errorf("create client plugin", "error", err.Error())
 	}
+	return err
+	//}, strategy.Wait(3*time.Second)); err != nil {
+	//	return fmt.Errorf("retry error to create plugin: %w", err)
+	//}
 
 	a.ibtpC = make(chan *pb.IBTP, IBTP_CH_SIZE)
 

internal/adapt/direct_adapter/adapter.go

@@ -1,6 +1,7 @@
 package direct_adapter
 
 import (
+	"context"
 	"fmt"
 	"runtime"
 	"sync"
@@ -33,6 +34,8 @@ type DirectAdapter struct {
 	ibtps           sync.Map
 	appchainID      string
 	remotePierID    string
+	ctx             context.Context
+	cancel          context.CancelFunc
 	gopool          *pool
 }
 
@@ -55,7 +58,7 @@ func (d *DirectAdapter) GetServiceIDList() ([]string, error) {
 func New(peerMgr peermgr.PeerManager, appchainAdapt adapt.Adapt, logger logrus.FieldLogger) (*DirectAdapter, error) {
 
 	appchainID := appchainAdapt.ID()
-
+	ctx, cancel := context.WithCancel(context.Background())
 	da := &DirectAdapter{
 		logger:        logger,
 		peerMgr:       peerMgr,
@@ -64,6 +67,8 @@ func New(peerMgr peermgr.PeerManager, appchainAdapt adapt.Adapt, logger logrus.F
 		ibtpC:         make(chan *pb.IBTP, maxChSize),
 		appchainID:    appchainID,
 		gopool:        NewGoPool(runtime.GOMAXPROCS(runtime.NumCPU())),
+		ctx:           ctx,
+		cancel:        cancel,
 	}
 
 	return da, nil

internal/adapt/direct_adapter/handler.go

@@ -82,31 +82,37 @@ func (d *DirectAdapter) handleSendIBTPMessage(stream network.Stream, msg *pb.Mes
 				}()
 				d.ibtpC <- ibtp
 				index := ibtp.Index
+				ticker := time.NewTicker(time.Second)
+				defer ticker.Stop()
 				for {
-					pool.lock.Lock()
-					if item := pool.ibtps.Min(); item != nil {
-						if item.(*MyTree).index < index+1 {
-							pool.ibtps.DeleteMin()
-						}
-
-						if item.(*MyTree).index == index+1 {
-							d.ibtpC <- item.(*MyTree).ibtp
-							pool.ibtps.DeleteMin()
-							index++
-							pool.time = time.Now()
-						}
-
-						// By default, the index will be equalized after 5 seconds
-						if time.Now().Sub(pool.time).Seconds() > 5.0 {
-							d.ibtpC <- item.(*MyTree).ibtp
-							pool.ibtps.DeleteMin()
-							index = item.(*MyTree).index
-							pool.time = time.Now()
+					select {
+					case <-ticker.C:
+						pool.lock.Lock()
+						if item := pool.ibtps.Min(); item != nil {
+							if item.(*MyTree).index < index+1 {
+								pool.ibtps.DeleteMin()
+							}
+
+							if item.(*MyTree).index == index+1 {
+								d.ibtpC <- item.(*MyTree).ibtp
+								pool.ibtps.DeleteMin()
+								index++
+								pool.time = time.Now()
+							}
+
+							// By default, the index will be equalized after 5 seconds
+							if time.Now().Sub(pool.time).Seconds() > 5.0 {
+								d.ibtpC <- item.(*MyTree).ibtp
+								pool.ibtps.DeleteMin()
+								index = item.(*MyTree).index
+								pool.time = time.Now()
+							}
 						}
+						pool.lock.Unlock()
+					case <-d.ctx.Done():
+						return
 					}
-					pool.lock.Unlock()
 				}
-
 			}(pool, ibtp)
 		} else {
 			pool.lock.Lock()