evm/cmd/simulator/txs/agent.go

// (c) 2023, Lux Industries, Inc. All rights reserved.
// See the file LICENSE for licensing terms.

package txs

import (
	"context"
	"errors"
	"fmt"
	"time"
	"github.com/luxfi/evm/cmd/simulator/metrics"
	"github.com/luxfi/geth/common"
	"github.com/luxfi/log"
)

type THash interface {
	Hash() common.Hash
}

// TxSequence provides an interface to return a channel of transactions.
// The sequence is responsible for closing the channel when there are no further
// transactions.
type TxSequence[T THash] interface {
	Chan() <-chan T
}

// Worker defines the interface for issuance and confirmation of transactions.
// The caller is responsible for calling Close to cleanup resources used by the
// worker at the end of the simulation.
type Worker[T THash] interface {
	IssueTx(ctx context.Context, tx T) error
	ConfirmTx(ctx context.Context, tx T) error
	LatestHeight(ctx context.Context) (uint64, error)
}

// Execute the work of the given agent.
type Agent[T THash] interface {
	Execute(ctx context.Context) error
}

// issueNAgent issues and confirms a batch of N transactions at a time.
type issueNAgent[T THash] struct {
	sequence TxSequence[T]
	worker   Worker[T]
	n        uint64
	metrics  *metrics.Metrics
}

// NewIssueNAgent creates a new issueNAgent
func NewIssueNAgent[T THash](sequence TxSequence[T], worker Worker[T], n uint64, metrics *metrics.Metrics) Agent[T] {
	return &issueNAgent[T]{
		sequence: sequence,
		worker:   worker,
		n:        n,
		metrics:  metrics,
	}
}

// Execute issues txs in batches of N and waits for them to confirm
func (a issueNAgent[T]) Execute(ctx context.Context) error {
	if a.n == 0 {
		return errors.New("batch size n cannot be equal to 0")
	}

	txChan := a.sequence.Chan()
	confirmedCount := 0
	batchI := 0
	m := a.metrics
	txMap := make(map[common.Hash]time.Time)

	// Tracks the total amount of time waiting for issuing and confirming txs
	var (
		totalIssuedTime    time.Duration
		totalConfirmedTime time.Duration
	)

	// Start time for execution
	start := time.Now()
	for {
		var (
			txs     = make([]T, 0, a.n)
			tx      T
			moreTxs bool
		)
		// Start issuance batch
		issuedStart := time.Now()
	L:
		for i := uint64(0); i < a.n; i++ {
			select {
			case <-ctx.Done():
				return ctx.Err()
			case tx, moreTxs = <-txChan:
				if !moreTxs {
					break L
				}
				issuanceIndividualStart := time.Now()
				txMap[tx.Hash()] = issuanceIndividualStart
				if err := a.worker.IssueTx(ctx, tx); err != nil {
					return fmt.Errorf("failed to issue transaction %d: %w", len(txs), err)
				}
				issuanceIndividualDuration := time.Since(issuanceIndividualStart)
				m.IssuanceTxTimes.Observe(issuanceIndividualDuration.Seconds())
				txs = append(txs, tx)
			}
		}
		// Get the batch's issuance time and add it to totalIssuedTime
		issuedDuration := time.Since(issuedStart)
		log.Info("Issuance Batch Done", "batch", batchI, "time", issuedDuration.Seconds())
		totalIssuedTime += issuedDuration

		// Wait for txs in this batch to confirm
		confirmedStart := time.Now()
		for i, tx := range txs {
			confirmedIndividualStart := time.Now()
			if err := a.worker.ConfirmTx(ctx, tx); err != nil {
				return fmt.Errorf("failed to await transaction %d: %w", i, err)
			}
			confirmationIndividualDuration := time.Since(confirmedIndividualStart)
			issuanceToConfirmationIndividualDuration := time.Since(txMap[tx.Hash()])
			m.ConfirmationTxTimes.Observe(confirmationIndividualDuration.Seconds())
			m.IssuanceToConfirmationTxTimes.Observe(issuanceToConfirmationIndividualDuration.Seconds())
			delete(txMap, tx.Hash())
			confirmedCount++
		}
		// Get the batch's confirmation time and add it to totalConfirmedTime
		confirmedDuration := time.Since(confirmedStart)
		log.Info("Confirmed Batch Done", "batch", batchI, "time", confirmedDuration.Seconds())
		totalConfirmedTime += confirmedDuration

		// Check if this is the last batch, if so write the final log and return
		if !moreTxs {
			totalTime := time.Since(start).Seconds()
			log.Info("Execution complete", "totalTxs", confirmedCount, "totalTime", totalTime, "TPS", float64(confirmedCount)/totalTime,
				"issuanceTime", totalIssuedTime.Seconds(), "confirmedTime", totalConfirmedTime.Seconds())

			return nil
		}

		batchI++
	}
}