worker upgraded

Author: pompon0

sender/sharded_sender.go

@@ -23,7 +23,6 @@ type ShardedSender struct {
 	mu         sync.RWMutex
 	collector  *stats.Collector
 	logger     *stats.Logger
-	limiter    *rate.Limiter // Shared rate limiter for all workers
 }
 
 // NewShardedSender creates a new sharded sender with workers for each endpoint
@@ -34,7 +33,7 @@ func NewShardedSender(cfg *config.LoadConfig, bufferSize int, workers int, limit
 
 	workerList := make([]*Worker, len(cfg.Endpoints))
 	for i, endpoint := range cfg.Endpoints {
-		workerList[i] = NewWorker(i, cfg.SeiChainID, endpoint, bufferSize, workers, limiter)
+		workerList[i] = NewWorker(i, cfg.SeiChainID, endpoint, bufferSize, workers)
 	}
 
 	return &ShardedSender{
@@ -46,13 +45,13 @@ func NewShardedSender(cfg *config.LoadConfig, bufferSize int, workers int, limit
 }
 
 // Start initializes and starts all workers
-func (s *ShardedSender) Run(ctx context.Context) error {
+func (s *ShardedSender) Run(ctx context.Context, limiter *rate.Limiter) error {
 	s.mu.Lock()
 	workers := s.workers
 	s.mu.Unlock()
 	return service.Run(ctx, func(ctx context.Context, s service.Scope) error {
 		for _, worker := range workers {
-			s.Spawn(func() error { return worker.Run(ctx) })
+			s.Spawn(func() error { return worker.Run(ctx,limiter) })
 		}
 		return nil
 	})

sender/worker.go

@@ -1,14 +1,10 @@
 package sender
 
 import (
-	"bytes"
 	"context"
 	"errors"
 	"fmt"
-	"io"
 	"log"
-	"net"
-	"net/http"
 	"time"
 
 	"github.com/ethereum/go-ethereum"
@@ -36,29 +32,10 @@ type Worker struct {
 	logger        *stats.Logger
 	workers       int
 	trackReceipts bool
-	limiter       *rate.Limiter // Shared rate limiter for transaction sending
-}
-
-func newHttpClient() *http.Client {
-	return &http.Client{
-		Timeout: 30 * time.Second,
-		Transport: &http.Transport{
-			DialContext: (&net.Dialer{
-				Timeout:   10 * time.Second,
-				KeepAlive: 30 * time.Second,
-			}).DialContext,
-			MaxIdleConns:          100,
-			MaxIdleConnsPerHost:   10,
-			IdleConnTimeout:       90 * time.Second,
-			TLSHandshakeTimeout:   10 * time.Second,
-			ExpectContinueTimeout: 1 * time.Second,
-			DisableKeepAlives:     false,
-		},
-	}
 }
 
 // NewWorker creates a new worker for a specific endpoint
-func NewWorker(id int, seiChainID string, endpoint string, bufferSize int, workers int, limiter *rate.Limiter) *Worker {
+func NewWorker(id int, seiChainID string, endpoint string, bufferSize int, workers int) *Worker {
 	w := &Worker{
 		id:            id,
 		seiChainID:    seiChainID,
@@ -67,7 +44,6 @@ func NewWorker(id int, seiChainID string, endpoint string, bufferSize int, worke
 		sentTxs:       make(chan *types.LoadTx, bufferSize),
 		workers:       workers,
 		trackReceipts: false,
-		limiter:       limiter,
 	}
 	meterWorkerQueueLength(w)
 	return w
@@ -80,12 +56,11 @@ func (w *Worker) SetStatsCollector(collector *stats.Collector, logger *stats.Log
 }
 
 // Start begins the worker's processing loop
-func (w *Worker) Run(ctx context.Context) error {
+func (w *Worker) Run(ctx context.Context, limiter *rate.Limiter) error {
 	return service.Run(ctx, func(ctx context.Context, s service.Scope) error {
 		// Start multiple worker goroutines that share the same channel
-		client := newHttpClient()
 		for range w.workers {
-			s.Spawn(func() error { return w.processTransactions(ctx, client) })
+			s.Spawn(func() error { return w.processTransactions(ctx,limiter) })
 		}
 		return w.watchTransactions(ctx)
 	})
@@ -115,18 +90,21 @@ func (w *Worker) watchTransactions(ctx context.Context) error {
 	if w.dryRun || !w.trackReceipts {
 		return nil
 	}
-	eth, err := ethclient.Dial(w.endpoint)
+
+	// Create a separate ethclient connection for receipt tracking
+	ethClient, err := ethclient.Dial(w.endpoint)
 	if err != nil {
 		return fmt.Errorf("ethclient.Dial(%q): %w", w.endpoint, err)
 	}
+	defer ethClient.Close()
 	for ctx.Err() == nil {
 		tx, err := utils.Recv(ctx, w.sentTxs)
 		if err != nil {
 			return err
 		}
 		ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
 		defer cancel()
-		if err := w.waitForReceipt(ctx, eth, tx); err != nil {
+		if err := w.waitForReceipt(ctx, ethClient, tx); err != nil {
 			log.Printf("❌ %v", err)
 		}
 	}
@@ -171,35 +149,40 @@ func (w *Worker) waitForReceipt(ctx context.Context, eth *ethclient.Client, tx *
 }
 
 // processTransactions is the main worker loop that processes transactions
-func (w *Worker) processTransactions(ctx context.Context, client *http.Client) error {
+func (w *Worker) processTransactions(ctx context.Context, limiter *rate.Limiter) error {
+	// Dial ethclient for this worker goroutine
+	ethClient, err := ethclient.Dial(w.endpoint)
+	if err != nil {
+		return fmt.Errorf("failed to connect to endpoint %s: %w", w.endpoint, err)
+	}
+	defer ethClient.Close()
+
 	for ctx.Err() == nil {
 		// Apply rate limiting before getting the next transaction
-		if w.limiter != nil {
-			if !w.limiter.Allow() {
-				continue
-			}
+		if err := limiter.Wait(ctx); err!=nil {
+			return err
 		}
-
 		tx, err := utils.Recv(ctx, w.txChan)
 		if err != nil {
 			return err
 		}
 
 		startTime := time.Now()
-		err = w.sendTransaction(ctx, client, tx)
+		// TODO: we cannot afford losing transactions due to nonce gaps.
+		// Consider retries though.
+		if err := w.sendTransaction(ctx, ethClient, tx); err != nil {
+			return fmt.Errorf("w.sendTransaction(): %w", err)
+		}
 		// Record statistics if collector is available
 		if w.collector != nil {
-			w.collector.RecordTransaction(tx.Scenario.Name, w.endpoint, time.Since(startTime), err == nil)
-		}
-		if err != nil {
-			log.Printf("%v", err)
+			w.collector.RecordTransaction(tx.Scenario.Name, w.endpoint, time.Since(startTime), false)
 		}
 	}
 	return ctx.Err()
 }
 
 // sendTransaction sends a single transaction to the endpoint
-func (w *Worker) sendTransaction(ctx context.Context, client *http.Client, tx *types.LoadTx) (_err error) {
+func (w *Worker) sendTransaction(ctx context.Context, ethClient *ethclient.Client, tx *types.LoadTx) (_err error) {
 	defer func(start time.Time) {
 		metrics.sendLatency.Record(ctx, time.Since(start).Seconds(),
 			metric.WithAttributes(
@@ -216,40 +199,11 @@ func (w *Worker) sendTransaction(ctx context.Context, client *http.Client, tx *t
 		return utils.Sleep(ctx, 10*time.Microsecond) // Much faster simulation
 	}
 
-	// Create HTTP request with JSON-RPC payload
-	req, err := http.NewRequestWithContext(ctx, "POST", w.endpoint, bytes.NewReader(tx.JSONRPCPayload))
-	if err != nil {
-		return fmt.Errorf("Worker %d: Failed to create request: %w", w.id, err)
-	}
-
-	// Set headers for JSON-RPC
-	req.Header.Set("Content-Type", "application/json")
-	req.Header.Set("Accept", "application/json")
-
-	// Send the request
-	resp, err := client.Do(req)
+	// Use go-ethereum client to send the transaction
+	err := ethClient.SendTransaction(ctx, tx.EthTx)
 	if err != nil {
 		return fmt.Errorf("Worker %d: Failed to send transaction: %w", w.id, err)
 	}
-	defer func() {
-		// Limit read to prevent memory issues with large responses
-		_, err = io.CopyN(io.Discard, resp.Body, 64*1024) // Read up to 64KB
-		if err != nil && err != io.EOF {
-			log.Printf("Worker %d: Failed to read response body: %v", w.id, err)
-			// Log but don't fail - this is just for connection reuse
-		}
-
-		// Close response body and handle error
-		if closeErr := resp.Body.Close(); closeErr != nil {
-			log.Printf("Worker %d: Failed to close response body: %v", w.id, closeErr)
-		}
-	}()
-
-	// Check response status
-	if resp.StatusCode != http.StatusOK {
-		return fmt.Errorf("Worker %d: HTTP error %d for transaction to %s", w.id, resp.StatusCode, w.endpoint)
-	}
-
 	// Write to sentTxs channel without blocking
 	select {
 	case w.sentTxs <- tx: