Arch refactor alpha (Hyperledger-TWGC#190)

* add crpto interface as repare for Hyperledger-TWGC#127 Signed-off-by: Sam Yuan <yy19902439@126.com> * adding Worker interface for Hyperledger-TWGC#56 and decouple Assembler and Integrator Signed-off-by: Sam Yuan <yy19902439@126.com> * refactor for worker interface Signed-off-by: Sam Yuan <yy19902439@126.com> * package structure Signed-off-by: Sam Yuan <yy19902439@126.com> * fix up Signed-off-by: Sam Yuan <yy19902439@126.com> * package refactor Signed-off-by: Sam Yuan <yy19902439@126.com> * fix up Signed-off-by: Sam Yuan <yy19902439@126.com> * fix up Signed-off-by: Sam Yuan <yy19902439@126.com> * remove Envelope from elements Signed-off-by: Sam Yuan <yy19902439@126.com> * add memeory free Signed-off-by: Sam Yuan <yy19902439@126.com> * remove Proposal from Elements Signed-off-by: Sam Yuan <yy19902439@126.com> * fix up Signed-off-by: Sam Yuan <yy19902439@126.com> * move start time to ctx Signed-off-by: Sam Yuan <yy19902439@126.com>
SamYuan1990 · Aug 24, 2021 · bd6f3f1 · bd6f3f1
1 parent f91227c
commit bd6f3f1
Show file tree

Hide file tree

Showing 4 changed files with 8 additions and 12 deletions.
diff --git a/pkg/infra/cmdImpl/process.go b/pkg/infra/cmdImpl/process.go
@@ -33,12 +33,13 @@ func Process(configPath string, num int, burst int, rate float64, logger *log.Lo
 	finishCh := make(chan struct{})
 	errorCh := make(chan error, burst)
 	ctx, cancel := context.WithCancel(context.Background())
+	ctx = context.WithValue(ctx, "start", time.Now())
 	defer cancel()
 	for i := 0; i < len(config.Endorsers); i++ {
 		signed[i] = make(chan *basic.Elements, burst)
 	}
 	/*** workers ***/
-	observer_workers, observers, err := observer.CreateObserverWorkers(config, crypto, blockCh, logger, ctx, finishCh, num, errorCh)
+	observer_workers, err := observer.CreateObserverWorkers(config, crypto, blockCh, logger, ctx, finishCh, num, errorCh)
 	if err != nil {
 		return err
 	}
@@ -59,7 +60,7 @@ func Process(configPath string, num int, burst int, rate float64, logger *log.Lo
 		case err = <-errorCh:
 			return err
 		case <-finishCh:
-			duration := time.Since(observers.GetTime())
+			duration := time.Since(ctx.Value("start").(time.Time))
 			logger.Infof("Completed processing transactions.")
 			fmt.Printf("tx: %d, duration: %+v, tps: %f\n", num, duration, float64(num)/duration.Seconds())
 			return nil

diff --git a/pkg/infra/observer/observer.go b/pkg/infra/observer/observer.go
@@ -47,16 +47,11 @@ func CreateObservers(ctx context.Context, crypto infra.Crypto, errorCh chan erro
 
 func (o *Observers) Start() {
 	//o.StartTime = time.Now()
-	o.ctx = context.WithValue(o.ctx, "start", time.Now())
 	for i := 0; i < len(o.workers); i++ {
 		go o.workers[i].Start(o.errorCh, o.blockCh, o.ctx.Value("start").(time.Time))
 	}
 }
 
-func (o *Observers) GetTime() time.Time {
-	return o.ctx.Value("start").(time.Time)
-}
-
 func CreateObserver(ctx context.Context, channel string, node basic.Node, crypto infra.Crypto, logger *log.Logger) (*Observer, error) {
 	seek, err := trafficGenerator.CreateSignedDeliverNewestEnv(channel, crypto)
 	if err != nil {

diff --git a/pkg/infra/observer/observerFactory.go b/pkg/infra/observer/observerFactory.go
@@ -9,17 +9,17 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
-func CreateObserverWorkers(config basic.Config, crypto infra.Crypto, blockCh chan *AddressedBlock, logger *log.Logger, ctx context.Context, finishCh chan struct{}, num int, errorCh chan error) ([]infra.Worker, *Observers, error) {
+func CreateObserverWorkers(config basic.Config, crypto infra.Crypto, blockCh chan *AddressedBlock, logger *log.Logger, ctx context.Context, finishCh chan struct{}, num int, errorCh chan error) ([]infra.Worker, error) {
 	observer_workers := make([]infra.Worker, 0)
 	blockCollector, err := NewBlockCollector(config.CommitThreshold, len(config.Committers), ctx, blockCh, finishCh, num, true)
 	if err != nil {
-		return observer_workers, nil, errors.Wrap(err, "failed to create block collector")
+		return observer_workers, errors.Wrap(err, "failed to create block collector")
 	}
 	observer_workers = append(observer_workers, blockCollector)
 	observers, err := CreateObservers(ctx, crypto, errorCh, blockCh, config, logger)
 	if err != nil {
-		return observer_workers, observers, err
+		return observer_workers, err
 	}
 	observer_workers = append(observer_workers, observers)
-	return observer_workers, observers, nil
+	return observer_workers, nil
 }
diff --git a/pkg/infra/trafficGenerator/proposal.go b/pkg/infra/trafficGenerator/proposal.go
@@ -26,7 +26,7 @@ const charset = "abcdefghijklmnopqrstuvwxyz" +
 var seededRand *rand.Rand = rand.New(
 	rand.NewSource(time.Now().UnixNano()))
 
-func CreateProposal(signer *Crypto, channel, ccname, version string, args ...string) (*peer.Proposal, error) {
+func CreateProposal(signer infra.Crypto, channel, ccname, version string, args ...string) (*peer.Proposal, error) {
 	var argsInByte [][]byte
 	for _, arg := range args {
 		// ref to https://ghz.sh/docs/calldata