Revert "feat: introduce encode stage (#814)"

This reverts commit 84cf1b58ae.

params/version.go

@@ -24,7 +24,7 @@ import (
 const (
 	VersionMajor = 5         // Major version component of the current release
 	VersionMinor = 4         // Minor version component of the current release
-	VersionPatch = 1         // Patch version component of the current release
+	VersionPatch = 0         // Patch version component of the current release
 	VersionMeta  = "mainnet" // Version metadata to append to the version string
 )
 

rollup/circuitcapacitychecker/impl.go

@@ -73,7 +73,19 @@ func (ccc *CircuitCapacityChecker) ApplyTransaction(traces *types.BlockTrace) (*
 	}
 
 	encodeStart := time.Now()
-	rustTrace := MakeRustTrace(traces, &ccc.jsonBuffer)
+	ccc.jsonBuffer.Reset()
+	err := json.NewEncoder(&ccc.jsonBuffer).Encode(traces)
+	if err != nil {
+		log.Error("fail to json marshal traces in ApplyTransaction", "id", ccc.ID, "TxHash", traces.Transactions[0].TxHash, "err", err)
+		return nil, ErrUnknown
+	}
+
+	tracesStr := C.CString(string(ccc.jsonBuffer.Bytes()))
+	defer func() {
+		C.free(unsafe.Pointer(tracesStr))
+	}()
+
+	rustTrace := C.parse_json_to_rust_trace(tracesStr)
 	if rustTrace == nil {
 		log.Error("fail to parse json in to rust trace", "id", ccc.ID, "TxHash", traces.Transactions[0].TxHash)
 		return nil, ErrUnknown
@@ -81,34 +93,26 @@ func (ccc *CircuitCapacityChecker) ApplyTransaction(traces *types.BlockTrace) (*
 	encodeTimer.UpdateSince(encodeStart)
 
 	log.Debug("start to check circuit capacity for tx", "id", ccc.ID, "TxHash", traces.Transactions[0].TxHash)
-	return ccc.applyTransactionRustTrace(rustTrace)
-}
-
-func (ccc *CircuitCapacityChecker) ApplyTransactionRustTrace(rustTrace unsafe.Pointer) (*types.RowConsumption, error) {
-	ccc.Lock()
-	defer ccc.Unlock()
-	return ccc.applyTransactionRustTrace(rustTrace)
-}
-
-func (ccc *CircuitCapacityChecker) applyTransactionRustTrace(rustTrace unsafe.Pointer) (*types.RowConsumption, error) {
 	rawResult := C.apply_tx(C.uint64_t(ccc.ID), rustTrace)
 	defer func() {
 		C.free_c_chars(rawResult)
 	}()
+	log.Debug("check circuit capacity for tx done", "id", ccc.ID, "TxHash", traces.Transactions[0].TxHash)
 
 	result := &WrappedRowUsage{}
-	if err := json.Unmarshal([]byte(C.GoString(rawResult)), result); err != nil {
+	if err = json.Unmarshal([]byte(C.GoString(rawResult)), result); err != nil {
-		log.Error("fail to json unmarshal apply_tx result", "id", ccc.ID, "err", err)
+		log.Error("fail to json unmarshal apply_tx result", "id", ccc.ID, "TxHash", traces.Transactions[0].TxHash, "err", err)
 		return nil, ErrUnknown
 	}
 
 	if result.Error != "" {
-		log.Error("fail to apply_tx in CircuitCapacityChecker", "id", ccc.ID, "err", result.Error)
+		log.Error("fail to apply_tx in CircuitCapacityChecker", "id", ccc.ID, "TxHash", traces.Transactions[0].TxHash, "err", result.Error)
 		return nil, ErrUnknown
 	}
 	if result.AccRowUsage == nil {
 		log.Error("fail to apply_tx in CircuitCapacityChecker",
-			"id", ccc.ID, "result.AccRowUsage == nil", result.AccRowUsage == nil,
+			"id", ccc.ID, "TxHash", traces.Transactions[0].TxHash,
+			"result.AccRowUsage == nil", result.AccRowUsage == nil,
 			"err", "AccRowUsage is empty unexpectedly")
 		return nil, ErrUnknown
 	}
@@ -124,7 +128,19 @@ func (ccc *CircuitCapacityChecker) ApplyBlock(traces *types.BlockTrace) (*types.
 	defer ccc.Unlock()
 
 	encodeStart := time.Now()
-	rustTrace := MakeRustTrace(traces, &ccc.jsonBuffer)
+	ccc.jsonBuffer.Reset()
+	err := json.NewEncoder(&ccc.jsonBuffer).Encode(traces)
+	if err != nil {
+		log.Error("fail to json marshal traces in ApplyBlock", "id", ccc.ID, "blockNumber", traces.Header.Number, "blockHash", traces.Header.Hash(), "err", err)
+		return nil, ErrUnknown
+	}
+
+	tracesStr := C.CString(string(ccc.jsonBuffer.Bytes()))
+	defer func() {
+		C.free(unsafe.Pointer(tracesStr))
+	}()
+
+	rustTrace := C.parse_json_to_rust_trace(tracesStr)
 	if rustTrace == nil {
 		log.Error("fail to parse json in to rust trace", "id", ccc.ID, "TxHash", traces.Transactions[0].TxHash)
 		return nil, ErrUnknown
@@ -139,7 +155,7 @@ func (ccc *CircuitCapacityChecker) ApplyBlock(traces *types.BlockTrace) (*types.
 	log.Debug("check circuit capacity for block done", "id", ccc.ID, "blockNumber", traces.Header.Number, "blockHash", traces.Header.Hash())
 
 	result := &WrappedRowUsage{}
-	if err := json.Unmarshal([]byte(C.GoString(rawResult)), result); err != nil {
+	if err = json.Unmarshal([]byte(C.GoString(rawResult)), result); err != nil {
 		log.Error("fail to json unmarshal apply_block result", "id", ccc.ID, "blockNumber", traces.Header.Number, "blockHash", traces.Header.Hash(), "err", err)
 		return nil, ErrUnknown
 	}
@@ -203,23 +219,3 @@ func (ccc *CircuitCapacityChecker) SetLightMode(lightMode bool) error {
 
 	return nil
 }
-
-func MakeRustTrace(trace *types.BlockTrace, buffer *bytes.Buffer) unsafe.Pointer {
-	if buffer == nil {
-		buffer = new(bytes.Buffer)
-	}
-	buffer.Reset()
-
-	err := json.NewEncoder(buffer).Encode(trace)
-	if err != nil {
-		log.Error("fail to json marshal traces in MakeRustTrace", "err", err)
-		return nil
-	}
-
-	tracesStr := C.CString(string(buffer.Bytes()))
-	defer func() {
-		C.free(unsafe.Pointer(tracesStr))
-	}()
-
-	return C.parse_json_to_rust_trace(tracesStr)
-}

rollup/circuitcapacitychecker/mock.go

@@ -3,9 +3,7 @@
 package circuitcapacitychecker
 
 import (
-	"bytes"
 	"math/rand"
-	"unsafe"
 
 	"github.com/scroll-tech/go-ethereum/common"
 	"github.com/scroll-tech/go-ethereum/core/types"
@@ -53,10 +51,6 @@ func (ccc *CircuitCapacityChecker) ApplyTransaction(traces *types.BlockTrace) (*
 	}}, nil
 }
 
-func (ccc *CircuitCapacityChecker) ApplyTransactionRustTrace(rustTrace unsafe.Pointer) (*types.RowConsumption, error) {
-	return ccc.ApplyTransaction(goTraces[rustTrace])
-}
-
 // ApplyBlock gets a block's RowConsumption.
 // Will only return a dummy value in mock_ccc.
 func (ccc *CircuitCapacityChecker) ApplyBlock(traces *types.BlockTrace) (*types.RowConsumption, error) {
@@ -88,11 +82,3 @@ func (ccc *CircuitCapacityChecker) Skip(txnHash common.Hash, err error) {
 	ccc.skipHash = txnHash.String()
 	ccc.skipError = err
 }
-
-var goTraces = make(map[unsafe.Pointer]*types.BlockTrace)
-
-func MakeRustTrace(trace *types.BlockTrace, buffer *bytes.Buffer) unsafe.Pointer {
-	rustTrace := new(struct{})
-	goTraces[unsafe.Pointer(rustTrace)] = trace
-	return unsafe.Pointer(rustTrace)
-}

rollup/pipeline/pipeline.go

@@ -1,12 +1,8 @@
 package pipeline
 
 import (
-	"bytes"
-	"context"
 	"errors"
-	"sync"
 	"time"
-	"unsafe"
 
 	"github.com/scroll-tech/go-ethereum/common"
 	"github.com/scroll-tech/go-ethereum/core"
@@ -34,7 +30,7 @@ func (e *ErrorWithTrace) Unwrap() error {
 }
 
 var (
-	ErrPipelineDone             = errors.New("pipeline is done")
+	ErrApplyStageDone           = errors.New("apply stage is done")
 	ErrUnexpectedL1MessageIndex = errors.New("unexpected L1 message index")
 
 	lifetimeTimer = func() metrics.Timer {
@@ -45,9 +41,6 @@ var (
 	applyTimer      = metrics.NewRegisteredTimer("pipeline/apply", nil)
 	applyIdleTimer  = metrics.NewRegisteredTimer("pipeline/apply_idle", nil)
 	applyStallTimer = metrics.NewRegisteredTimer("pipeline/apply_stall", nil)
-	encodeTimer      = metrics.NewRegisteredTimer("pipeline/encode", nil)
-	encodeIdleTimer  = metrics.NewRegisteredTimer("pipeline/encode_idle", nil)
-	encodeStallTimer = metrics.NewRegisteredTimer("pipeline/encode_stall", nil)
 	cccTimer        = metrics.NewRegisteredTimer("pipeline/ccc", nil)
 	cccIdleTimer    = metrics.NewRegisteredTimer("pipeline/ccc_idle", nil)
 )
@@ -57,9 +50,6 @@ type Pipeline struct {
 	vmConfig vm.Config
 	parent   *types.Block
 	start    time.Time
-	wg        sync.WaitGroup
-	ctx       context.Context
-	cancelCtx context.CancelFunc
 
 	// accumulators
 	ccc            *circuitcapacitychecker.CircuitCapacityChecker
@@ -93,8 +83,6 @@ func NewPipeline(
 	// make sure we are not sharing a tracer with the caller and not in debug mode
 	vmConfig.Tracer = nil
 	vmConfig.Debug = false
-
-	ctx, cancel := context.WithCancel(context.Background())
 	return &Pipeline{
 		chain:          chain,
 		vmConfig:       vmConfig,
@@ -104,8 +92,6 @@ func NewPipeline(
 		ccc:            ccc,
 		state:          state,
 		gasPool:        new(core.GasPool).AddGas(header.GasLimit),
-		ctx:            ctx,
-		cancelCtx:      cancel,
 	}
 }
 
@@ -117,20 +103,22 @@ func (p *Pipeline) WithBeforeTxHook(beforeTxHook func()) *Pipeline {
 func (p *Pipeline) Start(deadline time.Time) error {
 	p.start = time.Now()
 	p.txnQueue = make(chan *types.Transaction)
-	applyStageRespCh, applyToEncodeCh, err := p.traceAndApplyStage(p.txnQueue)
+	applyStageRespCh, candidateCh, err := p.traceAndApplyStage(p.txnQueue)
 	if err != nil {
 		log.Error("Failed starting traceAndApplyStage", "err", err)
 		return err
 	}
 	p.applyStageRespCh = applyStageRespCh
-	encodeToCccCh := p.encodeStage(applyToEncodeCh)
+	p.ResultCh = p.cccStage(candidateCh, deadline)
-	p.ResultCh = p.cccStage(encodeToCccCh, deadline)
 	return nil
 }
 
 // Stop forces pipeline to stop its operation and return whatever progress it has so far
 func (p *Pipeline) Stop() {
+	if p.txnQueue != nil {
 		close(p.txnQueue)
+		p.txnQueue = nil
+	}
 }
 
 func (p *Pipeline) TryPushTxns(txs types.OrderedTransactionSet, onFailingTxn func(txnIndex int, tx *types.Transaction, err error) bool) *Result {
@@ -149,7 +137,7 @@ func (p *Pipeline) TryPushTxns(txs types.OrderedTransactionSet, onFailingTxn fun
 		case err == nil, errors.Is(err, core.ErrNonceTooLow):
 			txs.Shift()
 		default:
-			if errors.Is(err, ErrPipelineDone) || onFailingTxn(p.txs.Len(), tx, err) {
+			if errors.Is(err, ErrApplyStageDone) || onFailingTxn(p.txs.Len(), tx, err) {
 				p.Stop()
 				return nil
 			}
@@ -166,11 +154,12 @@ func (p *Pipeline) TryPushTxns(txs types.OrderedTransactionSet, onFailingTxn fun
 }
 
 func (p *Pipeline) TryPushTxn(tx *types.Transaction) (*Result, error) {
+	if p.txnQueue == nil {
+		return nil, ErrApplyStageDone
+	}
 
 	select {
 	case p.txnQueue <- tx:
-	case <-p.ctx.Done():
-		return nil, ErrPipelineDone
 	case res := <-p.ResultCh:
 		return res, nil
 	}
@@ -178,7 +167,7 @@ func (p *Pipeline) TryPushTxn(tx *types.Transaction) (*Result, error) {
 	select {
 	case err, valid := <-p.applyStageRespCh:
 		if !valid {
-			return nil, ErrPipelineDone
+			return nil, ErrApplyStageDone
 		}
 		return nil, err
 	case res := <-p.ResultCh:
@@ -188,13 +177,18 @@ func (p *Pipeline) TryPushTxn(tx *types.Transaction) (*Result, error) {
 
 // Release releases all resources related to the pipeline
 func (p *Pipeline) Release() {
-	p.cancelCtx()
+	p.Stop()
-	p.wg.Wait()
+
+	select {
+	case <-p.applyStageRespCh:
+		<-p.ResultCh
+	case <-p.ResultCh:
+		<-p.applyStageRespCh
+	}
 }
 
 type BlockCandidate struct {
 	LastTrace      *types.BlockTrace
-	RustTrace      unsafe.Pointer
 	NextL1MsgIndex uint64
 
 	// accumulated state
@@ -205,47 +199,25 @@ type BlockCandidate struct {
 	CoalescedLogs []*types.Log
 }
 
-// sendCancellable tries to send msg to resCh but allows send operation to be cancelled
-// by closing cancelCh. Returns true if cancelled.
-func sendCancellable[T any, C comparable](resCh chan T, msg T, cancelCh <-chan C) bool {
-	var zeroC C
-
-	select {
-	case resCh <- msg:
-		return false
-	case cancelSignal := <-cancelCh:
-		if cancelSignal != zeroC {
-			panic("shouldn't have happened")
-		}
-		return true
-	}
-}
-
 func (p *Pipeline) traceAndApplyStage(txsIn <-chan *types.Transaction) (<-chan error, <-chan *BlockCandidate, error) {
 	p.state.StartPrefetcher("miner")
-	downstreamCh := make(chan *BlockCandidate, p.downstreamChCapacity())
+	newCandidateCh := make(chan *BlockCandidate)
 	resCh := make(chan error)
-	p.wg.Add(1)
 	go func() {
 		defer func() {
-			close(downstreamCh)
+			close(newCandidateCh)
 			close(resCh)
 			p.state.StopPrefetcher()
-			p.wg.Done()
 		}()
 
 		var tx *types.Transaction
 		for {
-			idleStart := time.Now()
+			applyIdleTimer.Time(func() {
-			select {
+				tx = <-txsIn
-			case tx = <-txsIn:
+			})
 			if tx == nil {
 				return
 			}
-			case <-p.ctx.Done():
-				return
-			}
-			applyIdleTimer.UpdateSince(idleStart)
 
 			applyStart := time.Now()
 
@@ -262,13 +234,13 @@ func (p *Pipeline) traceAndApplyStage(txsIn <-chan *types.Transaction) (<-chan e
 
 			if tx.IsL1MessageTx() && tx.AsL1MessageTx().QueueIndex != p.nextL1MsgIndex {
 				// Continue, we might still be able to include some L2 messages
-				sendCancellable(resCh, ErrUnexpectedL1MessageIndex, p.ctx.Done())
+				resCh <- ErrUnexpectedL1MessageIndex
 				continue
 			}
 
 			if !tx.IsL1MessageTx() && !p.chain.Config().Scroll.IsValidBlockSize(p.blockSize+tx.Size()) {
 				// can't fit this txn in this block, silently ignore and continue looking for more txns
-				sendCancellable(resCh, nil, p.ctx.Done())
+				resCh <- nil
 				continue
 			}
 
@@ -295,16 +267,21 @@ func (p *Pipeline) traceAndApplyStage(txsIn <-chan *types.Transaction) (<-chan e
 				}
 
 				stallStart := time.Now()
-				if sendCancellable(downstreamCh, &BlockCandidate{
+				select {
-					NextL1MsgIndex: p.nextL1MsgIndex,
+				case newCandidateCh <- &BlockCandidate{
 					LastTrace:      trace,
+					NextL1MsgIndex: p.nextL1MsgIndex,
 
 					Header:        types.CopyHeader(&p.Header),
 					State:         p.state.Copy(),
 					Txs:           p.txs,
 					Receipts:      p.receipts,
 					CoalescedLogs: p.coalescedLogs,
-				}, p.ctx.Done()) {
+				}:
+				case tx = <-txsIn:
+					if tx != nil {
+						panic("shouldn't have happened")
+					}
 					// next stage terminated and caller terminated us as well
 					return
 				}
@@ -317,10 +294,10 @@ func (p *Pipeline) traceAndApplyStage(txsIn <-chan *types.Transaction) (<-chan e
 				}
 			}
 			applyTimer.UpdateSince(applyStart)
-			sendCancellable(resCh, err, p.ctx.Done())
+			resCh <- err
 		}
 	}()
-	return resCh, downstreamCh, nil
+	return resCh, newCandidateCh, nil
 }
 
 type Result struct {
@@ -332,70 +309,25 @@ type Result struct {
 	FinalBlock *BlockCandidate
 }
 
-func (p *Pipeline) encodeStage(traces <-chan *BlockCandidate) <-chan *BlockCandidate {
-	downstreamCh := make(chan *BlockCandidate, p.downstreamChCapacity())
-	p.wg.Add(1)
-
-	go func() {
-		defer func() {
-			close(downstreamCh)
-			p.wg.Done()
-		}()
-		buffer := new(bytes.Buffer)
-		for {
-			idleStart := time.Now()
-			select {
-			case trace := <-traces:
-				if trace == nil {
-					return
-				}
-				encodeIdleTimer.UpdateSince(idleStart)
-
-				encodeStart := time.Now()
-				if p.ccc != nil {
-					trace.RustTrace = circuitcapacitychecker.MakeRustTrace(trace.LastTrace, buffer)
-					if trace.RustTrace == nil {
-						log.Error("making rust trace", "txHash", trace.LastTrace.Transactions[0].TxHash)
-						return
-					}
-				}
-				encodeTimer.UpdateSince(encodeStart)
-
-				stallStart := time.Now()
-				sendCancellable(downstreamCh, trace, p.ctx.Done())
-				encodeStallTimer.UpdateSince(stallStart)
-			case <-p.ctx.Done():
-				return
-			}
-
-		}
-	}()
-	return downstreamCh
-}
-
 func (p *Pipeline) cccStage(candidates <-chan *BlockCandidate, deadline time.Time) <-chan *Result {
 	if p.ccc != nil {
 		p.ccc.Reset()
 	}
-	resultCh := make(chan *Result, 1)
+	resultCh := make(chan *Result)
 	var lastCandidate *BlockCandidate
 	var lastAccRows *types.RowConsumption
 	var deadlineReached bool
 
-	p.wg.Add(1)
 	go func() {
 		deadlineTimer := time.NewTimer(time.Until(deadline))
 		defer func() {
 			close(resultCh)
 			deadlineTimer.Stop()
 			lifetimeTimer.UpdateSince(p.start)
-			p.wg.Done()
 		}()
 		for {
 			idleStart := time.Now()
 			select {
-			case <-p.ctx.Done():
-				return
 			case <-deadlineTimer.C:
 				cccIdleTimer.UpdateSince(idleStart)
 				// note: currently we don't allow empty blocks, but if we ever do; make sure to CCC check it first
@@ -413,7 +345,7 @@ func (p *Pipeline) cccStage(candidates <-chan *BlockCandidate, deadline time.Tim
 				var accRows *types.RowConsumption
 				var err error
 				if candidate != nil && p.ccc != nil {
-					accRows, err = p.ccc.ApplyTransactionRustTrace(candidate.RustTrace)
+					accRows, err = p.ccc.ApplyTransaction(candidate.LastTrace)
 					lastTxn := candidate.Txs[candidate.Txs.Len()-1]
 					cccTimer.UpdateSince(cccStart)
 					if err != nil {
@@ -495,13 +427,3 @@ func (p *Pipeline) traceAndApply(tx *types.Transaction) (*types.Receipt, *types.
 	}
 	return receipt, trace, nil
 }
-
-// downstreamChCapacity returns the channel capacity that should be used for downstream channels.
-// It aims to minimize stalls caused by different computational costs of different transactions
-func (p *Pipeline) downstreamChCapacity() int {
-	cap := 1
-	if p.chain.Config().Scroll.MaxTxPerBlock != nil {
-		cap = *p.chain.Config().Scroll.MaxTxPerBlock
-	}
-	return cap
-}