From 146b18ad15e45b84cfd0ac69bbeb5307611f3915 Mon Sep 17 00:00:00 2001 From: smcio Date: Sun, 26 Jul 2026 09:39:35 +0200 Subject: [PATCH] prebuild and slim dependency planner --- pkg/metrics/metrics.go | 14 +- pkg/replay/block.go | 182 ++++---- pkg/replay/block_preparation_test.go | 312 ++++++++++++-- pkg/replay/topsort_planner.go | 624 ++++++++++++++++++++------- pkg/replay/topsort_planner_test.go | 286 ++++++++++++ pkg/replay/transaction.go | 18 - scripts/replay_timings_viewer.py | 13 +- 7 files changed, 1151 insertions(+), 298 deletions(-) diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go index 39f91a997..3ac452b2f 100644 --- a/pkg/metrics/metrics.go +++ b/pkg/metrics/metrics.go @@ -142,11 +142,19 @@ type BlockReplay struct { DependencyPlannerPreparation Timing LoadBlockAccounts Timing SlotCtxSetup Timing - // DependencyPlannerBuild is the graph/batch construction time nested - // within TxLoop. DependencyPlannerDispatch is the scheduler's wall-clock - // lifetime, including dependency waits, and is also nested within TxLoop. + // DependencyPlannerBuild is measured account-extraction plus graph/batch + // construction work, independent of which planner route a block uses and + // excluding goroutine scheduling delay. On the prepared route some or all + // of it overlaps LoadBlockAccounts and it is therefore not a top-level + // additive phase. DependencyPlannerWait is the residual join nested within + // TxLoop. DependencyPlannerDispatch runs from a ready plan until its final + // transaction wave is enqueued, excluding the final wave's execution, and + // is also nested within TxLoop. DependencyPlannerBuild Timing + DependencyPlannerWait Timing DependencyPlannerDispatch Timing + DependencyPlannerPrepared uint64 + DependencyPlannerFallback uint64 TxLoop Timing Reward Timing Rent Timing diff --git a/pkg/replay/block.go b/pkg/replay/block.go index cc369abc9..1ed27f0ea 100644 --- a/pkg/replay/block.go +++ b/pkg/replay/block.go @@ -530,7 +530,13 @@ func recordSysvarAccountReadStats(dst *metrics.AccountLoader, src accountsdb.Acc } } -func loadBlockAccountsAndUpdateSysvars(accountsDb blockAccountSource, block *b.Block, epochSchedule *sealevel.SysvarEpochSchedule, alpenglowClock bool) (accounts.Accounts, accounts.Accounts, int, error) { +func loadBlockAccountsAndUpdateSysvars( + accountsDb blockAccountSource, + block *b.Block, + epochSchedule *sealevel.SysvarEpochSchedule, + alpenglowClock bool, + planner *preparedDependencyPlanner, +) (accounts.Accounts, accounts.Accounts, int, error) { phaseStart := time.Now() err := resolveAddrTableLookups(accountsDb, block) metrics.GlobalBlockReplay.AccountLoader.AddressTableLookups.AddTimingSince(phaseStart) @@ -538,6 +544,15 @@ func loadBlockAccountsAndUpdateSysvars(accountsDb blockAccountSource, block *b.B return nil, nil, 0, err } + // Live ALT transactions have no RPC metadata, so their account accesses are + // knowable only after lookup resolution. Extract their compact accesses and + // build the graph concurrently with account loading and sysvar updates. For + // RPC blocks this is a no-op because preparation began safely from static + // keys plus TransactionMeta before resolution mutated the execution messages. + if planner != nil { + planner.tryStartResolved(block) + } + phaseStart = time.Now() dedupedAccts, uniqueWritableAccounts := extractAndDedupeBlockAccts(block) publicationCapacity := publicationMapCapacity(block, uniqueWritableAccounts, alpenglowClock) @@ -3522,10 +3537,14 @@ func lightbringerEntryExecutionBatches(transactions []*solana.Transaction, entry } segmentBatches[batchIdx] = append(segmentBatches[batchIdx], txIdx) for _, roAcct := range readonlyAccounts { - lastReadBatch[roAcct] = batchIdx + if previous, exists := lastReadBatch[roAcct]; !exists || batchIdx > previous { + lastReadBatch[roAcct] = batchIdx + } } for _, writeAcct := range writableAccounts { - lastWriteBatch[writeAcct] = batchIdx + if previous, exists := lastWriteBatch[writeAcct]; !exists || batchIdx > previous { + lastWriteBatch[writeAcct] = batchIdx + } } } *batches = append(*batches, segmentBatches...) @@ -3551,28 +3570,35 @@ func lightbringerEntryExecutionBatches(transactions []*solana.Transaction, entry return batches } -func parallelTxLoop(slotCtx *sealevel.SlotCtx, sigverifyWg *sync.WaitGroup, block *b.Block, rblock *b.Block, executionPlan blockTransactionExecutionPlan, txParallelism int, dbgOpts *DebugOptions, shouldVerifySignatures bool) (fees.TxFeeInfoAccumulator, uint64) { +func parallelTxLoop( + slotCtx *sealevel.SlotCtx, + sigverifyWg *sync.WaitGroup, + planner *preparedDependencyPlanner, + block *b.Block, + executionPlan blockTransactionExecutionPlan, + txParallelism int, + dbgOpts *DebugOptions, + shouldVerifySignatures bool, +) (fees.TxFeeInfoAccumulator, uint64) { var txFeeAccumulator fees.TxFeeInfoAccumulator txFeeInfos := make([]*fees.TxFeeInfo, len(block.Transactions)) txComputeUnitsConsumed := make([]uint64, len(block.Transactions)) errs := make([]error, len(block.Transactions)) - plannerBlock := block - if rblock.FromLiveStream { - plannerBlock = rblock - } - - if canUseDependencyPlanner(plannerBlock) { + plannerWaitStart := time.Now() + dependencyPlan, plannerBuildDuration, plannerAvailable := planner.wait() + metrics.GlobalBlockReplay.DependencyPlannerWait.AddTimingSince(plannerWaitStart) + if plannerAvailable { + metrics.GlobalBlockReplay.DependencyPlannerPrepared = 1 + metrics.GlobalBlockReplay.DependencyPlannerBuild.AddTiming(plannerBuildDuration) do := make(chan int, len(block.Transactions)) done := make(chan int, len(block.Transactions)) plannerDone := make(chan struct{}) go func() { defer close(plannerDone) - plannerStart := time.Now() - topsortPlannerStream(plannerBlock, do, done, func() { - metrics.GlobalBlockReplay.DependencyPlannerBuild.AddTimingSince(plannerStart) - }) - metrics.GlobalBlockReplay.DependencyPlannerDispatch.AddTimingSince(plannerStart) + plannerDispatchStart := time.Now() + dispatchDependencyPlan(dependencyPlan, do, done) + metrics.GlobalBlockReplay.DependencyPlannerDispatch.AddTimingSince(plannerDispatchStart) }() wg := &sync.WaitGroup{} @@ -3587,10 +3613,10 @@ func parallelTxLoop(slotCtx *sealevel.SlotCtx, sigverifyWg *sync.WaitGroup, bloc } tx := block.Transactions[idx] var txMeta *rpc.TransactionMeta - if idx < len(rblock.TxMetas) { - txMeta = rblock.TxMetas[idx] + if idx < len(block.TxMetas) { + txMeta = block.TxMetas[idx] } - txFeeInfos[idx], txComputeUnitsConsumed[idx], errs[idx] = ProcessTransaction(slotCtx, sigverifyWg, rblock.Transactions[idx], txMeta, dbgOpts, sealevel.BorrowedAccountArenas[i], shouldVerifySignatures) + txFeeInfos[idx], txComputeUnitsConsumed[idx], errs[idx] = ProcessTransaction(slotCtx, sigverifyWg, block.Transactions[idx], txMeta, dbgOpts, sealevel.BorrowedAccountArenas[i], shouldVerifySignatures) txErr := errs[idx] // check for success-failure return value divergences if txMeta != nil && txErr == nil && txMeta.Err != nil { @@ -3611,12 +3637,29 @@ func parallelTxLoop(slotCtx *sealevel.SlotCtx, sigverifyWg *sync.WaitGroup, bloc wg.Wait() close(done) <-plannerDone - } else if rblock.FromLiveStream { - plannerDispatchStart := time.Now() - var plannerBuildDuration time.Duration + } else if block.FromLiveStream { + metrics.GlobalBlockReplay.DependencyPlannerFallback = 1 + // Include any unsuccessful prepared-planner attempt in the total planner + // work, then build all fallback batches before timing dispatch. This + // keeps Build and Dispatch comparable across the two planner modes. + fallbackBuildStart := time.Now() + var executionBatches [][]uint64 + relaxIntraBatchAccountLocks := block.Features != nil && + block.Features.IsActive(features.RelaxIntraBatchAccountLocks) + for _, entry := range block.Entries { + executionBatches = append( + executionBatches, + lightbringerEntryExecutionBatches(block.Transactions, entry, relaxIntraBatchAccountLocks)..., + ) + } + plannerBuildDuration += time.Since(fallbackBuildStart) + metrics.GlobalBlockReplay.DependencyPlannerBuild.AddTiming(plannerBuildDuration) + batchWg := &sync.WaitGroup{} workersWg := &sync.WaitGroup{} - do := make(chan uint64, txParallelism) + // The full-block buffer makes final-wave enqueue non-blocking, matching + // the prepared CSR dispatch timer's end boundary. + do := make(chan uint64, len(block.Transactions)) workersWg.Add(txParallelism) for i := range txParallelism { go func(workerIdx int) { @@ -3624,10 +3667,10 @@ func parallelTxLoop(slotCtx *sealevel.SlotCtx, sigverifyWg *sync.WaitGroup, bloc for idx := range do { tx := block.Transactions[idx] var txMeta *rpc.TransactionMeta - if int(idx) < len(rblock.TxMetas) { - txMeta = rblock.TxMetas[idx] + if int(idx) < len(block.TxMetas) { + txMeta = block.TxMetas[idx] } - txFeeInfos[idx], txComputeUnitsConsumed[idx], errs[idx] = ProcessTransaction(slotCtx, sigverifyWg, rblock.Transactions[idx], txMeta, dbgOpts, sealevel.BorrowedAccountArenas[workerIdx], shouldVerifySignatures) + txFeeInfos[idx], txComputeUnitsConsumed[idx], errs[idx] = ProcessTransaction(slotCtx, sigverifyWg, block.Transactions[idx], txMeta, dbgOpts, sealevel.BorrowedAccountArenas[workerIdx], shouldVerifySignatures) txErr := errs[idx] if txMeta != nil && txErr == nil && txMeta.Err != nil { mlog.Log.Errorf("[run:%s] DIVERGENCE in slot %d: tx %s succeeded locally but failed onchain: %+v", @@ -3643,32 +3686,30 @@ func parallelTxLoop(slotCtx *sealevel.SlotCtx, sigverifyWg *sync.WaitGroup, bloc }(i) } - relaxIntraBatchAccountLocks := rblock.Features != nil && - rblock.Features.IsActive(features.RelaxIntraBatchAccountLocks) - for _, entry := range rblock.Entries { - plannerBuildStart := time.Now() - batches := lightbringerEntryExecutionBatches(rblock.Transactions, entry, relaxIntraBatchAccountLocks) - plannerBuildDuration += time.Since(plannerBuildStart) - for _, batch := range batches { - executable := 0 - for _, txIdx := range batch { - if executionPlan.execute[txIdx] { - executable++ - } + plannerDispatchStart := time.Now() + for batchIdx, batch := range executionBatches { + executable := 0 + for _, txIdx := range batch { + if executionPlan.execute[txIdx] { + executable++ } - batchWg.Add(executable) - for _, txIdx := range batch { - if executionPlan.execute[txIdx] { - do <- txIdx - } + } + batchWg.Add(executable) + for _, txIdx := range batch { + if executionPlan.execute[txIdx] { + do <- txIdx } - batchWg.Wait() } + if batchIdx == len(executionBatches)-1 { + metrics.GlobalBlockReplay.DependencyPlannerDispatch.AddTimingSince(plannerDispatchStart) + } + batchWg.Wait() + } + if len(executionBatches) == 0 { + metrics.GlobalBlockReplay.DependencyPlannerDispatch.AddTimingSince(plannerDispatchStart) } close(do) workersWg.Wait() - metrics.GlobalBlockReplay.DependencyPlannerBuild.AddTiming(plannerBuildDuration) - metrics.GlobalBlockReplay.DependencyPlannerDispatch.AddTimingSince(plannerDispatchStart) } else { panic("dependency planner unavailable for non-Lightbringer block") } @@ -3700,39 +3741,6 @@ func parallelTxLoop(slotCtx *sealevel.SlotCtx, sigverifyWg *sync.WaitGroup, bloc return txFeeAccumulator, totalComputeUnitsConsumed } -// prepareDependencyPlannerBlock preserves unresolved transaction account keys -// only when the dependency planner needs them. Live replay explicitly plans -// from the execution block after address-table resolution, and sequential -// replay has no planner, so cloning either kind would be pure overhead. -func prepareDependencyPlannerBlock(block *b.Block, txParallelism int) (*b.Block, error) { - if block == nil { - return nil, errors.New("nil block") - } - if txParallelism <= 0 || block.FromLiveStream { - return block, nil - } - - unresolvedBlock := &b.Block{ - Transactions: make([]*solana.Transaction, len(block.Transactions)), - TxMetas: make([]*rpc.TransactionMeta, len(block.TxMetas)), - Slot: block.Slot, - ParentSlot: block.ParentSlot, - } - for i := range block.Transactions { - clonedTx, err := cloneTransaction(block.Transactions[i]) - if err != nil { - return nil, fmt.Errorf("clone transaction %d in slot %d: %w", i, block.Slot, err) - } - unresolvedBlock.Transactions[i] = clonedTx - if i < len(block.TxMetas) && block.TxMetas[i] != nil { - unresolvedBlock.TxMetas[i] = &rpc.TransactionMeta{} - *unresolvedBlock.TxMetas[i] = *block.TxMetas[i] - } - } - - return unresolvedBlock, nil -} - func ProcessBlock( acctsDb *accountsdb.AccountsDb, block *b.Block, @@ -3775,7 +3783,7 @@ func ProcessBlock( replayStage.Store(stage) replayStageSince.Store(time.Now().UnixNano()) } - setReplayStage("clone_transactions") + setReplayStage("prepare_dependency_planner") replayWatchdogDone := make(chan struct{}) go func() { @@ -3825,11 +3833,15 @@ func ProcessBlock( metrics.GlobalBlockReplay.SignatureVerificationJoin.AddTimingSince(sigverifyJoinStart) }() plannerPreparationStart := time.Now() - plannerBlock, err := prepareDependencyPlannerBlock(block, txParallelism) - metrics.GlobalBlockReplay.DependencyPlannerPreparation.AddTimingSince(plannerPreparationStart) - if err != nil { - panic(fmt.Sprintf("unable to prepare dependency planner block for slot %d: %v", block.Slot, err)) + var planner *preparedDependencyPlanner + if txParallelism > 0 { + planner = newPreparedDependencyPlanner() + // RPC metadata makes unresolved ALT accesses available before lookup + // resolution. Live blocks without metadata start preparation immediately + // after resolution in loadBlockAccountsAndUpdateSysvars. + planner.tryStart(block) } + metrics.GlobalBlockReplay.DependencyPlannerPreparation.AddTimingSince(plannerPreparationStart) start := time.Now() setReplayStage("load_accounts") @@ -3840,7 +3852,7 @@ func ProcessBlock( if tail != nil { blockSrc = tail } - accts, parentAccts, accountMapCapacity, err := loadBlockAccountsAndUpdateSysvars(blockSrc, block, epochSchedule, alpenglowClock) + accts, parentAccts, accountMapCapacity, err := loadBlockAccountsAndUpdateSysvars(blockSrc, block, epochSchedule, alpenglowClock, planner) loadAcctsRegion.End() if err != nil { panic(fmt.Sprintf("unable to load slot accounts and update sysvars: %s", err)) @@ -3860,7 +3872,7 @@ func ProcessBlock( txLoopRegion := trace.StartRegion(ctx, "TxLoop") shouldVerifySignatures := !block.TransactionSignaturesVerified() if txParallelism > 0 { - txFeeAccumulator, totalComputeUnitsConsumed = parallelTxLoop(slotCtx, &sigverifyWg, plannerBlock, block, executionPlan, txParallelism, dbgOpts, shouldVerifySignatures) + txFeeAccumulator, totalComputeUnitsConsumed = parallelTxLoop(slotCtx, &sigverifyWg, planner, block, executionPlan, txParallelism, dbgOpts, shouldVerifySignatures) } else { txFeeAccumulator, totalComputeUnitsConsumed = sequentialTxLoop(slotCtx, &sigverifyWg, block, executionPlan, dbgOpts, shouldVerifySignatures) } diff --git a/pkg/replay/block_preparation_test.go b/pkg/replay/block_preparation_test.go index ff019e897..4a4ed6ccb 100644 --- a/pkg/replay/block_preparation_test.go +++ b/pkg/replay/block_preparation_test.go @@ -4,56 +4,302 @@ import ( "testing" b "github.com/Overclock-Validator/mithril/pkg/block" - "github.com/Overclock-Validator/mithril/pkg/tpu/txfixture" "github.com/gagliardetto/solana-go" "github.com/gagliardetto/solana-go/rpc" ) -func TestPrepareDependencyPlannerBlockSkipsUnneededClone(t *testing.T) { - tx, err := solana.TransactionFromBytes(txfixture.MustSignedTransferWire(1)) - if err != nil { - t.Fatalf("decode fixture transaction: %v", err) +func unresolvedPlannerTestTransaction(payer, table solana.PublicKey) *solana.Transaction { + msg := solana.Message{ + Header: solana.MessageHeader{ + NumRequiredSignatures: 1, + }, + AccountKeys: []solana.PublicKey{payer}, + AddressTableLookups: solana.MessageAddressTableLookupSlice{ + { + AccountKey: table, + WritableIndexes: []uint8{0}, + }, + }, } + msg.SetVersion(solana.MessageVersionV0) + return &solana.Transaction{Message: msg} +} + +func TestPreparedDependencyPlannerSnapshotsUnresolvedMetadata(t *testing.T) { + sharedWritable := solana.PublicKey{0x40} + block := &b.Block{ + Slot: 99, + Transactions: []*solana.Transaction{ + unresolvedPlannerTestTransaction(solana.PublicKey{0x10}, solana.PublicKey{0x20}), + unresolvedPlannerTestTransaction(solana.PublicKey{0x11}, solana.PublicKey{0x21}), + }, + TxMetas: []*rpc.TransactionMeta{ + {LoadedAddresses: rpc.LoadedAddresses{Writable: []solana.PublicKey{sharedWritable}}}, + {LoadedAddresses: rpc.LoadedAddresses{Writable: []solana.PublicKey{sharedWritable}}}, + }, + } + + planner := newPreparedDependencyPlanner() + if !planner.tryStart(block) { + t.Fatal("planner did not start from unresolved transactions with metadata") + } + + // Lookup resolution mutates messages, and callers may release RPC metadata + // once preparation returns. Neither may alter the already snapshotted plan. + block.Transactions[1].Message.AccountKeys[0] = solana.PublicKey{0x50} + block.TxMetas[1].LoadedAddresses.Writable[0] = solana.PublicKey{0x51} + + plan, _, available := planner.wait() + if !available { + t.Fatal("prepared plan was unavailable") + } + if got := len(plan.dependents); got != 1 { + t.Fatalf("dependency edges = %d, want 1", got) + } + if plan.dependents[0] != 1 || plan.offsets[0] != 0 || plan.offsets[1] != 1 { + t.Fatalf("unexpected prepared dependency plan: offsets=%v dependents=%v", plan.offsets, plan.dependents) + } +} + +func TestPreparedDependencyPlannerStartsAfterResolution(t *testing.T) { + sharedWritable := solana.PublicKey{0x70} + transaction := func(payer solana.PublicKey) *solana.Transaction { + return &solana.Transaction{ + Message: solana.Message{ + Header: solana.MessageHeader{ + NumRequiredSignatures: 1, + }, + AccountKeys: []solana.PublicKey{payer, sharedWritable}, + }, + } + } + block := &b.Block{ + Transactions: []*solana.Transaction{ + transaction(solana.PublicKey{0x71}), + transaction(solana.PublicKey{0x72}), + }, + } + + planner := newPreparedDependencyPlanner() + planner.tryStartResolved(block) + plan, _, available := planner.wait() + if !available { + t.Fatal("planner unavailable for derivable resolved block") + } + if got := len(plan.dependents); got != 1 { + t.Fatalf("dependency edges = %d, want 1", got) + } +} +func TestPreparedDependencyPlannerPreservesUnresolvedLiveFallback(t *testing.T) { + block := &b.Block{ + FromLiveStream: true, + Transactions: []*solana.Transaction{ + unresolvedPlannerTestTransaction(solana.PublicKey{0x80}, solana.PublicKey{0x81}), + }, + } + + planner := newPreparedDependencyPlanner() + planner.tryStartResolved(block) + if _, _, available := planner.wait(); available { + t.Fatal("planner unexpectedly available for unresolved live transaction without metadata") + } +} + +func TestPreparedDependencyPlannerRejectsIncompleteLookupMetadata(t *testing.T) { for _, test := range []struct { - name string - live bool - parallelism int - wantSameBlock bool + name string + txMeta *rpc.TransactionMeta }{ - {name: "live parallel", live: true, parallelism: 4, wantSameBlock: true}, - {name: "live sequential", live: true, parallelism: 0, wantSameBlock: true}, - {name: "rpc sequential", live: false, parallelism: 0, wantSameBlock: true}, - {name: "rpc parallel", live: false, parallelism: 4, wantSameBlock: false}, + {name: "missing loaded address", txMeta: &rpc.TransactionMeta{}}, + { + name: "wrong writable split", + txMeta: &rpc.TransactionMeta{ + LoadedAddresses: rpc.LoadedAddresses{ + ReadOnly: []solana.PublicKey{{0x92}}, + }, + }, + }, } { t.Run(test.name, func(t *testing.T) { block := &b.Block{ - Slot: 99, - FromLiveStream: test.live, - Transactions: []*solana.Transaction{tx}, - TxMetas: []*rpc.TransactionMeta{{Fee: 5}}, + Transactions: []*solana.Transaction{ + unresolvedPlannerTestTransaction(solana.PublicKey{0x90}, solana.PublicKey{0x91}), + }, + TxMetas: []*rpc.TransactionMeta{test.txMeta}, + } + + planner := newPreparedDependencyPlanner() + if planner.tryStart(block) { + t.Fatal("planner started from incomplete or misclassified lookup metadata") + } + }) + } +} + +func fallbackBatchTestTransaction( + writableAccounts []solana.PublicKey, + readonlyAccounts []solana.PublicKey, +) *solana.Transaction { + accountKeys := make([]solana.PublicKey, 0, len(writableAccounts)+len(readonlyAccounts)) + accountKeys = append(accountKeys, writableAccounts...) + accountKeys = append(accountKeys, readonlyAccounts...) + return &solana.Transaction{ + Message: solana.Message{ + Header: solana.MessageHeader{ + NumReadonlyUnsignedAccounts: uint8(len(readonlyAccounts)), + }, + AccountKeys: accountKeys, + }, + } +} + +func TestLightbringerFallbackRetainsHighestReadBatch(t *testing.T) { + accountX := solana.PublicKey{0xa0} + accountA := solana.PublicKey{0xa1} + transactions := []*solana.Transaction{ + fallbackBatchTestTransaction([]solana.PublicKey{accountX}, nil), + fallbackBatchTestTransaction(nil, []solana.PublicKey{accountX, accountA}), + fallbackBatchTestTransaction(nil, []solana.PublicKey{accountA}), + fallbackBatchTestTransaction([]solana.PublicKey{accountA}, nil), + } + entry := &b.TxEntry{Indices: []uint64{0, 1, 2, 3}} + + batches := lightbringerEntryExecutionBatches(transactions, entry, true) + want := [][]uint64{{0, 2}, {1}, {3}} + if len(batches) != len(want) { + t.Fatalf("batches = %v, want %v", batches, want) + } + for idx := range want { + if len(batches[idx]) != len(want[idx]) { + t.Fatalf("batches = %v, want %v", batches, want) + } + for txIdx := range want[idx] { + if batches[idx][txIdx] != want[idx][txIdx] { + t.Fatalf("batches = %v, want %v", batches, want) } - prepared, err := prepareDependencyPlannerBlock(block, test.parallelism) - if err != nil { - t.Fatalf("prepare block: %v", err) + } + } +} + +func FuzzLightbringerFallbackPreservesConflictOrder(f *testing.F) { + // The first seed encodes the regression covered above: a high-batch reader + // of account A is followed by a lower-batch reader, then a writer of A. + f.Add([]byte{ + 3, + 0, 2, 1, + 1, 2, 0, 1, 0, + 0, 1, 0, + 0, 1, 1, + }) + f.Add([]byte{7, 3, 1, 1, 2, 0, 3, 1, 2, 1, 4, 0}) + f.Add([]byte{15, 7, 5, 0, 5, 1, 6, 0, 7, 1, 8, 0}) + + f.Fuzz(func(t *testing.T, data []byte) { + if len(data) < 2 { + t.Skip() + } + + cursor := 1 + nextByte := func() byte { + value := data[cursor%len(data)] + cursor++ + return value + } + + transactionCount := 1 + int(data[0]%64) + transactions := make([]*solana.Transaction, transactionCount) + indices := make([]uint64, transactionCount) + // 0 means absent, 1 means read-only, and 2 means writable. Repeated + // accesses are canonicalized by promoting the account to writable. + accountModes := make([][32]uint8, transactionCount) + for txIdx := range transactions { + indices[txIdx] = uint64(txIdx) + accessCount := 1 + int(nextByte()%8) + for range accessCount { + accountIdx := nextByte() % 32 + mode := uint8(1) + if nextByte()&1 != 0 { + mode = 2 + } + if mode > accountModes[txIdx][accountIdx] { + accountModes[txIdx][accountIdx] = mode + } } - if gotSame := prepared == block; gotSame != test.wantSameBlock { - t.Fatalf("same block = %t, want %t", gotSame, test.wantSameBlock) + + payer := solana.PublicKey{0xff, byte(txIdx)} + accountKeys := []solana.PublicKey{payer} + numReadonly := 0 + for accountIdx, mode := range accountModes[txIdx] { + if mode == 2 { + accountKeys = append(accountKeys, solana.PublicKey{byte(accountIdx)}) + } + } + for accountIdx, mode := range accountModes[txIdx] { + if mode == 1 { + accountKeys = append(accountKeys, solana.PublicKey{byte(accountIdx)}) + numReadonly++ + } } - if test.wantSameBlock { - return + transactions[txIdx] = &solana.Transaction{ + Message: solana.Message{ + Header: solana.MessageHeader{ + NumRequiredSignatures: 1, + NumReadonlyUnsignedAccounts: uint8(numReadonly), + }, + AccountKeys: accountKeys, + }, } - if prepared.Transactions[0] == block.Transactions[0] { - t.Fatal("parallel RPC replay did not clone its unresolved transaction") + } + + batches := lightbringerEntryExecutionBatches( + transactions, + &b.TxEntry{Indices: indices}, + true, + ) + batchOf := make([]int, transactionCount) + seen := make([]bool, transactionCount) + for batchIdx, batch := range batches { + for _, rawTxIdx := range batch { + if rawTxIdx >= uint64(transactionCount) { + t.Fatalf("batch contains out-of-range transaction %d", rawTxIdx) + } + txIdx := int(rawTxIdx) + if seen[txIdx] { + t.Fatalf("transaction %d appears in more than one batch", txIdx) + } + seen[txIdx] = true + batchOf[txIdx] = batchIdx } - if prepared.TxMetas[0] == block.TxMetas[0] { - t.Fatal("parallel RPC replay did not copy its transaction metadata") + } + for txIdx, wasSeen := range seen { + if !wasSeen { + t.Fatalf("transaction %d is absent from fallback batches", txIdx) } + } - block.Transactions[0].Message.RecentBlockhash[0] ^= 0xff - if prepared.Transactions[0].Message.RecentBlockhash == block.Transactions[0].Message.RecentBlockhash { - t.Fatal("prepared transaction changed when the resolved transaction was mutated") + conflicts := func(earlier, later int) bool { + for accountIdx, earlierMode := range accountModes[earlier] { + laterMode := accountModes[later][accountIdx] + if earlierMode != 0 && laterMode != 0 && + (earlierMode == 2 || laterMode == 2) { + return true + } } - }) - } + return false + } + for earlier := 0; earlier < transactionCount; earlier++ { + for later := earlier + 1; later < transactionCount; later++ { + if conflicts(earlier, later) && batchOf[earlier] >= batchOf[later] { + t.Fatalf( + "conflicting transactions %d and %d have batches %d and %d", + earlier, + later, + batchOf[earlier], + batchOf[later], + ) + } + } + } + }) } diff --git a/pkg/replay/topsort_planner.go b/pkg/replay/topsort_planner.go index da5b9db62..3d0238d58 100644 --- a/pkg/replay/topsort_planner.go +++ b/pkg/replay/topsort_planner.go @@ -2,9 +2,8 @@ package replay import ( "fmt" - //"time" + "time" - //"github.com/Overclock-Validator/mithril/pkg/mlog" "github.com/Overclock-Validator/mithril/pkg/block" "github.com/Overclock-Validator/mithril/pkg/util" "github.com/gagliardetto/solana-go" @@ -12,7 +11,6 @@ import ( ) // Type wrappers around indices. -type acct int type tx int func canDeriveAccountsFromMessage(t *solana.Transaction) bool { @@ -76,176 +74,479 @@ func messageReadonlyAccounts(msg *solana.Message) []solana.PublicKey { return util.DedupePubkeys(accounts) } -func getAllAccounts(t *solana.Transaction, tm *rpc.TransactionMeta) []solana.PublicKey { - if tm == nil && canDeriveAccountsFromMessage(t) { - return util.DedupePubkeys(append([]solana.PublicKey(nil), t.Message.AccountKeys...)) - } - if tm == nil { - panic("unresolved transaction requires txmeta to derive all accounts") +func canUseDependencyPlanner(b *block.Block) bool { + for i, tx := range b.Transactions { + if i < len(b.TxMetas) && plannerShouldUseMeta(tx, b.TxMetas[i]) { + continue + } + if canDeriveAccountsFromMessage(tx) { + continue + } + return false } + return true +} - accounts := make([]solana.PublicKey, 0, len(t.Message.AccountKeys)+len(tm.LoadedAddresses.ReadOnly)+len(tm.LoadedAddresses.Writable)) - accounts = append(accounts, t.Message.AccountKeys...) - accounts = append(accounts, tm.LoadedAddresses.ReadOnly...) - accounts = append(accounts, tm.LoadedAddresses.Writable...) - return util.DedupePubkeys(accounts) +// plannerTransactionAccounts is the planner's immutable, compact view of one +// transaction. Accounts before readonlyEnd are read-only; the remainder are +// writable. Keeping only public keys and one boundary avoids preserving a full +// transaction (and its metadata) while address-table resolution mutates the +// execution block. +type plannerTransactionAccounts struct { + accounts []plannerAccountAccess + readonlyEnd int } -func getReadonlyAccounts(t *solana.Transaction, tm *rpc.TransactionMeta) []solana.PublicKey { - if tm == nil && canDeriveAccountsFromMessage(t) { - return messageReadonlyAccounts(&t.Message) - } - if tm == nil { - panic("unresolved transaction requires txmeta to derive readonly accounts") - } - msg := t.Message - hdr := msg.Header - numReadonly := len(tm.LoadedAddresses.ReadOnly) - signedReadonly := int(hdr.NumReadonlySignedAccounts) - numReadonly += signedReadonly - unsignedReadonly := int(hdr.NumReadonlyUnsignedAccounts) - numReadonly += unsignedReadonly - accounts := make([]solana.PublicKey, 0, numReadonly) +type plannerAccountAccess struct { + publicKey solana.PublicKey + writable bool +} - accounts = append(accounts, msg.AccountKeys[int(hdr.NumRequiredSignatures)-signedReadonly:hdr.NumRequiredSignatures]...) - accounts = append(accounts, msg.AccountKeys[len(msg.AccountKeys)-unsignedReadonly:len(msg.AccountKeys)]...) - accounts = append(accounts, tm.LoadedAddresses.ReadOnly...) - return util.DedupePubkeys(accounts) +func (a plannerTransactionAccounts) readonly() []plannerAccountAccess { + return a.accounts[:a.readonlyEnd] } -func getWritableAccounts(t *solana.Transaction, tm *rpc.TransactionMeta) []solana.PublicKey { - if tm == nil && canDeriveAccountsFromMessage(t) { - return messageWritableAccounts(&t.Message) - } - if tm == nil { - panic("unresolved transaction requires txmeta to derive writable accounts") +func (a plannerTransactionAccounts) writable() []plannerAccountAccess { + return a.accounts[a.readonlyEnd:] +} + +type plannerTransactionAccountBuilder struct { + accounts []plannerAccountAccess + seen map[solana.PublicKey]int +} + +const plannerLinearDedupeLimit = 32 + +func newPlannerTransactionAccountBuilder(capacity int) plannerTransactionAccountBuilder { + return plannerTransactionAccountBuilder{ + accounts: make([]plannerAccountAccess, 0, capacity), } - msg := t.Message - hdr := msg.Header - numWritable := len(tm.LoadedAddresses.Writable) - signedWritable := int(hdr.NumRequiredSignatures) - int(hdr.NumReadonlySignedAccounts) - numWritable += signedWritable - unsignedWritable := len(msg.AccountKeys) - int(hdr.NumRequiredSignatures) - int(hdr.NumReadonlyUnsignedAccounts) - numWritable += unsignedWritable - accounts := make([]solana.PublicKey, 0, numWritable) +} - accounts = append(accounts, msg.AccountKeys[0:signedWritable]...) - accounts = append(accounts, msg.AccountKeys[int(hdr.NumRequiredSignatures):int(hdr.NumRequiredSignatures)+unsignedWritable]...) - accounts = append(accounts, tm.LoadedAddresses.Writable...) - return util.DedupePubkeys(accounts) +func plannerTransactionAccountBuilderFromStorage(storage []plannerAccountAccess) plannerTransactionAccountBuilder { + return plannerTransactionAccountBuilder{accounts: storage[:0]} } -func canUseDependencyPlanner(b *block.Block) bool { - for i, tx := range b.Transactions { - if i < len(b.TxMetas) && b.TxMetas[i] != nil { +// add deduplicates accounts and promotes any key encountered as both read-only +// and writable to writable. A linear scan is intentionally cheaper than one +// hash map allocation per transaction for Solana's small sanitized key lists. +func (b *plannerTransactionAccountBuilder) add(account solana.PublicKey, writable bool) { + if b.seen != nil { + if idx, exists := b.seen[account]; exists { + b.accounts[idx].writable = b.accounts[idx].writable || writable + return + } + b.seen[account] = len(b.accounts) + b.accounts = append(b.accounts, plannerAccountAccess{ + publicKey: account, + writable: writable, + }) + return + } + for idx := range b.accounts { + if b.accounts[idx].publicKey != account { continue } - if canDeriveAccountsFromMessage(tx) { + b.accounts[idx].writable = b.accounts[idx].writable || writable + return + } + if len(b.accounts) == plannerLinearDedupeLimit { + b.seen = make(map[solana.PublicKey]int, cap(b.accounts)) + for idx := range b.accounts { + b.seen[b.accounts[idx].publicKey] = idx + } + b.seen[account] = len(b.accounts) + } + b.accounts = append(b.accounts, plannerAccountAccess{ + publicKey: account, + writable: writable, + }) +} + +func (b *plannerTransactionAccountBuilder) finish() plannerTransactionAccounts { + // Partition in place so the compact view needs only one backing array. + // Account order does not affect the dependency relation. + readonlyEnd := 0 + for idx := range b.accounts { + if b.accounts[idx].writable { continue } + b.accounts[readonlyEnd], b.accounts[idx] = b.accounts[idx], b.accounts[readonlyEnd] + readonlyEnd++ + } + return plannerTransactionAccounts{ + accounts: b.accounts, + readonlyEnd: readonlyEnd, + } +} + +// newPlannerTransactionAccounts is the compact test/convenience constructor. +func newPlannerTransactionAccounts( + capacity int, + addAccounts func(add func(solana.PublicKey, bool)), +) plannerTransactionAccounts { + builder := newPlannerTransactionAccountBuilder(capacity) + addAccounts(builder.add) + return builder.finish() +} + +func addPlannerAccountsFromMessage(builder *plannerTransactionAccountBuilder, msg *solana.Message) { + numStaticAccounts, numWritableLookupAccounts := messageAccountLayout(msg) + for idx, account := range msg.AccountKeys { + builder.add(account, messageAccountIsWritable(msg, idx, numStaticAccounts, numWritableLookupAccounts)) + } +} + +func plannerAccountsFromMessage(msg *solana.Message) plannerTransactionAccounts { + builder := newPlannerTransactionAccountBuilder(len(msg.AccountKeys)) + addPlannerAccountsFromMessage(&builder, msg) + return builder.finish() +} + +func plannerStaticAccountCount(msg *solana.Message) int { + numStaticAccounts := len(msg.AccountKeys) + if msg.IsResolved() { + numStaticAccounts -= msg.NumLookups() + } + if numStaticAccounts < 0 { + panic("resolved transaction has more lookup accounts than account keys") + } + return numStaticAccounts +} + +func addPlannerAccountsFromMeta( + builder *plannerTransactionAccountBuilder, + t *solana.Transaction, + tm *rpc.TransactionMeta, +) { + msg := &t.Message + numStaticAccounts := plannerStaticAccountCount(msg) + requiredSignatures := int(msg.Header.NumRequiredSignatures) + signedWritableEnd := requiredSignatures - int(msg.Header.NumReadonlySignedAccounts) + unsignedWritableEnd := numStaticAccounts - int(msg.Header.NumReadonlyUnsignedAccounts) + if signedWritableEnd < 0 || requiredSignatures > numStaticAccounts || + unsignedWritableEnd < requiredSignatures { + panic("transaction message header account ranges are invalid") + } + + for idx, account := range msg.AccountKeys[:numStaticAccounts] { + isWritable := idx < signedWritableEnd || + (idx >= requiredSignatures && idx < unsignedWritableEnd) + builder.add(account, isWritable) + } + for _, account := range tm.LoadedAddresses.ReadOnly { + builder.add(account, false) + } + for _, account := range tm.LoadedAddresses.Writable { + builder.add(account, true) + } +} + +func plannerAccountsFromMeta(t *solana.Transaction, tm *rpc.TransactionMeta) plannerTransactionAccounts { + capacity := plannerStaticAccountCount(&t.Message) + len(tm.LoadedAddresses.ReadOnly) + len(tm.LoadedAddresses.Writable) + builder := newPlannerTransactionAccountBuilder(capacity) + addPlannerAccountsFromMeta(&builder, t, tm) + return builder.finish() +} + +func plannerShouldUseMeta(t *solana.Transaction, tm *rpc.TransactionMeta) bool { + // Once lookup resolution has populated the message, it is the canonical + // complete account layout. Metadata is needed only to snapshot unresolved + // v0 lookup accounts before resolution mutates their messages. A nonnil but + // incomplete metadata object is not sufficient: defer preparation until + // the message has been resolved instead of silently omitting account locks. + if tm == nil || t.Message.IsResolved() { return false } - return true + lookupCount := t.Message.NumLookups() + if !t.Message.IsVersioned() || lookupCount == 0 { + return true + } + writableLookupCount := t.Message.GetAddressTableLookups().NumWritableLookups() + readonlyLookupCount := lookupCount - writableLookupCount + return len(tm.LoadedAddresses.Writable) == writableLookupCount && + len(tm.LoadedAddresses.ReadOnly) == readonlyLookupCount } -func blockToDependencyGraph(b *block.Block) (adjacencyList [][]tx, inDegree []int) { - //start := time.Now() - // Map between pubkeys and account indices - var acctToPk []solana.PublicKey - pkToAcct := make(map[solana.PublicKey]acct, len(b.Transactions)*4) +func plannerAccountsForBlock(b *block.Block) ([]plannerTransactionAccounts, bool) { + if b == nil || !canUseDependencyPlanner(b) { + return nil, false + } - for i, tx := range b.Transactions { + // One shared backing allocation replaces one account-slice allocation per + // transaction. Each transaction receives a disjoint capacity-bounded window + // and is compacted independently within it. + totalCapacity := 0 + for idx, transaction := range b.Transactions { + capacity := len(transaction.Message.AccountKeys) + if idx < len(b.TxMetas) && plannerShouldUseMeta(transaction, b.TxMetas[idx]) { + txMeta := b.TxMetas[idx] + capacity = plannerStaticAccountCount(&transaction.Message) + + len(txMeta.LoadedAddresses.ReadOnly) + len(txMeta.LoadedAddresses.Writable) + } + totalCapacity += capacity + } + accountStorage := make([]plannerAccountAccess, totalCapacity) + accounts := make([]plannerTransactionAccounts, len(b.Transactions)) + storageOffset := 0 + for idx, transaction := range b.Transactions { var txMeta *rpc.TransactionMeta - if i < len(b.TxMetas) { - txMeta = b.TxMetas[i] + if idx < len(b.TxMetas) { + txMeta = b.TxMetas[idx] } - - accounts := getAllAccounts(tx, txMeta) - for _, acctPk := range accounts { - if _, exists := pkToAcct[acctPk]; !exists { - pkToAcct[acctPk] = acct(len(acctToPk)) - acctToPk = append(acctToPk, acctPk) - } + if !plannerShouldUseMeta(transaction, txMeta) { + txMeta = nil } + capacity := len(transaction.Message.AccountKeys) + if txMeta != nil { + capacity = plannerStaticAccountCount(&transaction.Message) + + len(txMeta.LoadedAddresses.ReadOnly) + len(txMeta.LoadedAddresses.Writable) + } + builder := plannerTransactionAccountBuilderFromStorage( + accountStorage[storageOffset : storageOffset : storageOffset+capacity], + ) + if txMeta != nil { + addPlannerAccountsFromMeta(&builder, transaction, txMeta) + } else { + addPlannerAccountsFromMessage(&builder, &transaction.Message) + } + accounts[idx] = builder.finish() + storageOffset += capacity } + return accounts, true +} - acctToReaderTxs := make(map[acct][]tx, len(acctToPk)) - acctToWriterTxs := make(map[acct][]tx, len(acctToPk)) - adjacencyList = make([][]tx, len(b.Transactions)) - inDegree = make([]int, len(b.Transactions)) - for txIdx := range b.Transactions { - /* Given S < T (S occurs before T) in a sequential execution, we - use these restrictions on the parallel execution to reproduce - the sequential execution - - S | T | parallel ordering required - reads | reads | any order - writes | reads | S < T - reads | writes | S < T - writes | writes | S < T - */ - t := tx(txIdx) - var txMeta *rpc.TransactionMeta - if txIdx < len(b.TxMetas) { - txMeta = b.TxMetas[txIdx] +type dependencyPlan struct { + // offsets/dependents are a compact sparse-row adjacency list. Transaction + // i's dependents are dependents[offsets[i]:offsets[i+1]]. + offsets []uint32 + dependents []uint32 + inDegree []uint32 +} + +type dependencyAccountState struct { + lastWriterPlusOne uint32 + firstReaderPlusOne uint32 + moreReaders []uint32 +} + +type dependencyEdge struct { + prerequisite uint32 + dependent uint32 +} + +func addDependency( + edges *[]dependencyEdge, + inDegree, outDegree, seen []uint32, + dependent, prerequisite uint32, +) { + generation := dependent + 1 + if seen[prerequisite] == generation { + return + } + seen[prerequisite] = generation + *edges = append(*edges, dependencyEdge{prerequisite: prerequisite, dependent: dependent}) + inDegree[dependent]++ + outDegree[prerequisite]++ +} + +// buildDependencyPlan creates a reachability-equivalent reduction of the old +// all-conflicts graph. A read depends only on the latest writer. A write +// depends on that writer and every reader since it. Older conflicts remain +// ordered transitively, which preserves sequential block semantics while +// avoiding quadratic writer chains and retaining far fewer edges. +func buildDependencyPlan(transactions []plannerTransactionAccounts) *dependencyPlan { + if uint64(len(transactions)) > uint64(^uint32(0)) { + panic("dependency planner transaction count exceeds uint32") + } + inDegree := make([]uint32, len(transactions)) + outDegree := make([]uint32, len(transactions)) + edges := make([]dependencyEdge, 0, len(transactions)*4) + accountToState := make(map[solana.PublicKey]int, len(transactions)*4) + accountStates := make([]dependencyAccountState, 0, len(transactions)*4) + seenDependencies := make([]uint32, len(transactions)) + + stateFor := func(account solana.PublicKey) *dependencyAccountState { + stateIdx, exists := accountToState[account] + if !exists { + stateIdx = len(accountStates) + accountToState[account] = stateIdx + accountStates = append(accountStates, dependencyAccountState{}) } + return &accountStates[stateIdx] + } - //txSig := b.Transactions[txIdx].Signatures[0] - ////mlog.Log.Debugf("printing input accounts for txIdx=%d txSig=%s", txIdx, txSig) - readonlyAccounts := getReadonlyAccounts(b.Transactions[txIdx], txMeta) - for _, roAcct := range readonlyAccounts { - ////mlog.Log.Debugf("- roAcct=%s", roAcct.String()) - acct, exists := pkToAcct[roAcct] - if !exists { - panic(fmt.Sprintf("invariant error: did not record account index for pk=%s in previous loop?", roAcct.String())) + for txIdx, transaction := range transactions { + currentTx := uint32(txIdx) + for _, account := range transaction.readonly() { + state := stateFor(account.publicKey) + if state.lastWriterPlusOne != 0 { + addDependency(&edges, inDegree, outDegree, seenDependencies, currentTx, state.lastWriterPlusOne-1) } - - // Add an edge for S writes < T reads - for _, s := range acctToWriterTxs[acct] { - if s >= t { - break - } - if len(adjacencyList[int(s)]) > 0 && adjacencyList[int(s)][len(adjacencyList[int(s)])-1] == t { - continue - } - adjacencyList[int(s)] = append(adjacencyList[int(s)], t) - inDegree[int(t)]++ + if state.firstReaderPlusOne == 0 { + state.firstReaderPlusOne = currentTx + 1 + } else { + state.moreReaders = append(state.moreReaders, currentTx) } - // Add T as a reader of this account. - acctToReaderTxs[acct] = append(acctToReaderTxs[acct], t) } - - writableAccts := getWritableAccounts(b.Transactions[txIdx], txMeta) - for _, writeAcct := range writableAccts { - ////mlog.Log.Debugf("- writeAcct=%s", writeAcct.String()) - acct, exists := pkToAcct[writeAcct] - if !exists { - panic(fmt.Sprintf("invariant error: expected pkToAcct to contain all public keys of accounts used in block; missing public key=%s", writeAcct.String())) + for _, account := range transaction.writable() { + state := stateFor(account.publicKey) + if state.lastWriterPlusOne != 0 { + addDependency(&edges, inDegree, outDegree, seenDependencies, currentTx, state.lastWriterPlusOne-1) } - // Add an edge for S reads < T writes - for _, s := range acctToReaderTxs[acct] { - if s >= t { - break - } - if len(adjacencyList[int(s)]) > 0 && adjacencyList[int(s)][len(adjacencyList[int(s)])-1] == t { - continue - } - adjacencyList[int(s)] = append(adjacencyList[int(s)], t) - inDegree[int(t)]++ + if state.firstReaderPlusOne != 0 { + addDependency(&edges, inDegree, outDegree, seenDependencies, currentTx, state.firstReaderPlusOne-1) } - // Add an edge for S writes < T writes - for _, s := range acctToWriterTxs[acct] { - if s >= t { - break - } - if len(adjacencyList[int(s)]) > 0 && adjacencyList[int(s)][len(adjacencyList[int(s)])-1] == t { - continue - } - adjacencyList[int(s)] = append(adjacencyList[int(s)], t) - inDegree[int(t)]++ + for _, reader := range state.moreReaders { + addDependency(&edges, inDegree, outDegree, seenDependencies, currentTx, reader) } - // Add T as a writer of this account. - acctToWriterTxs[acct] = append(acctToWriterTxs[acct], t) + state.firstReaderPlusOne = 0 + state.moreReaders = state.moreReaders[:0] + state.lastWriterPlusOne = currentTx + 1 + } + } + + offsets := make([]uint32, len(transactions)+1) + var edgeCount uint64 + for idx, count := range outDegree { + edgeCount += uint64(count) + if edgeCount > uint64(^uint32(0)) { + panic("dependency planner edge count exceeds uint32") + } + offsets[idx+1] = uint32(edgeCount) + } + dependents := make([]uint32, len(edges)) + nextOffset := append([]uint32(nil), offsets[:len(transactions)]...) + for _, edge := range edges { + offset := nextOffset[edge.prerequisite] + dependents[offset] = edge.dependent + nextOffset[edge.prerequisite]++ + } + return &dependencyPlan{ + offsets: offsets, + dependents: dependents, + inDegree: inDegree, + } +} + +func dependencyPlanForBlock(b *block.Block) (*dependencyPlan, bool) { + accounts, available := plannerAccountsForBlock(b) + if !available { + return nil, false + } + return buildDependencyPlan(accounts), true +} + +// preparedDependencyPlanner owns a plan built ahead of TxLoop. Before lookup +// resolution it snapshots compact account accesses synchronously; afterward it +// can safely overlap extraction and graph construction with account loading. +type preparedDependencyPlanner struct { + started bool + result chan preparedDependencyPlanResult + ready *preparedDependencyPlanResult +} + +type preparedDependencyPlanResult struct { + plan *dependencyPlan + buildDuration time.Duration + available bool +} + +func newPreparedDependencyPlanner() *preparedDependencyPlanner { + return &preparedDependencyPlanner{} +} + +func plannerBlockHasAddressTableLookups(b *block.Block) bool { + for _, transaction := range b.Transactions { + if transaction.Message.IsVersioned() && transaction.Message.NumLookups() > 0 { + return true + } + } + return false +} + +func (p *preparedDependencyPlanner) tryStart(b *block.Block) bool { + if p == nil || p.started { + return p != nil && p.started + } + if b == nil || !canUseDependencyPlanner(b) { + return false + } + if !plannerBlockHasAddressTableLookups(b) { + // Lookup resolution cannot mutate these messages, so both compact + // extraction and graph construction may overlap account loading. + p.tryStartResolved(b) + return true + } + accountExtractionStart := time.Now() + accounts, available := plannerAccountsForBlock(b) + if !available { + return false + } + accountExtractionDuration := time.Since(accountExtractionStart) + p.started = true + p.result = make(chan preparedDependencyPlanResult, 1) + go func() { + graphBuildStart := time.Now() + plan := buildDependencyPlan(accounts) + p.result <- preparedDependencyPlanResult{ + plan: plan, + buildDuration: accountExtractionDuration + time.Since(graphBuildStart), + available: true, + } + }() + return true +} + +// tryStartResolved may read the block in the builder goroutine because lookup +// resolution has completed and transaction messages are immutable from this +// point onward. This overlaps both compact account extraction and graph +// construction with the rest of account loading. +func (p *preparedDependencyPlanner) tryStartResolved(b *block.Block) { + if p == nil || p.started { + return + } + p.started = true + p.result = make(chan preparedDependencyPlanResult, 1) + go func() { + buildStart := time.Now() + accounts, available := plannerAccountsForBlock(b) + var plan *dependencyPlan + if available { + plan = buildDependencyPlan(accounts) + } + p.result <- preparedDependencyPlanResult{ + plan: plan, + buildDuration: time.Since(buildStart), + available: available, + } + }() +} + +func (p *preparedDependencyPlanner) wait() (*dependencyPlan, time.Duration, bool) { + if p == nil || !p.started { + return nil, 0, false + } + if p.ready == nil { + result := <-p.result + p.ready = &result + } + return p.ready.plan, p.ready.buildDuration, p.ready.available +} + +func blockToDependencyGraph(b *block.Block) (adjacencyList [][]tx, inDegree []int) { + plan, available := dependencyPlanForBlock(b) + if !available { + panic(fmt.Sprintf("dependency planner cannot derive accounts for block at slot %d", b.Slot)) + } + adjacencyList = make([][]tx, len(plan.inDegree)) + inDegree = make([]int, len(plan.inDegree)) + for idx := range plan.inDegree { + inDegree[idx] = int(plan.inDegree[idx]) + start, end := plan.offsets[idx], plan.offsets[idx+1] + adjacencyList[idx] = make([]tx, end-start) + for edgeIdx, dependent := range plan.dependents[start:end] { + adjacencyList[idx][edgeIdx] = tx(dependent) } } return adjacencyList, inDegree @@ -255,13 +556,16 @@ func blockToDependencyGraph(b *block.Block) (adjacencyList [][]tx, inDegree []in // The ints are indices into the b.Transactions slices. // Each list of indices do not have write-after-write or read-after-write conflicts. func TopsortPlanner(b *block.Block) [][]int { - adjList, inDegree := blockToDependencyGraph(b) + plan, available := dependencyPlanForBlock(b) + if !available { + panic(fmt.Sprintf("dependency planner cannot derive accounts for block at slot %d", b.Slot)) + } // Output a topological sorting of the transactions topSorted := 0 var topSortLevels [][]int var roots []int - for t, deg := range inDegree { + for t, deg := range plan.inDegree { if deg == 0 { roots = append(roots, t) } @@ -272,9 +576,10 @@ func TopsortPlanner(b *block.Block) [][]int { // Remove roots from graph. var nextRoots []int for _, root := range roots { - for _, dependentTx := range adjList[int(root)] { - inDegree[int(dependentTx)]-- - if inDegree[int(dependentTx)] == 0 { + start, end := plan.offsets[root], plan.offsets[root+1] + for _, dependentTx := range plan.dependents[start:end] { + plan.inDegree[dependentTx]-- + if plan.inDegree[dependentTx] == 0 { nextRoots = append(nextRoots, int(dependentTx)) } } @@ -294,25 +599,34 @@ func TopsortPlannerStream(b *block.Block, out chan int, done chan int) { // topsortPlannerStream exposes the graph-build boundary to replay metrics // without coupling the generally useful planner to the global collector. func topsortPlannerStream(b *block.Block, out chan int, done chan int, onGraphBuilt func()) { - adjList, inDegree := blockToDependencyGraph(b) + plan, available := dependencyPlanForBlock(b) + if !available { + panic(fmt.Sprintf("dependency planner cannot derive accounts for block at slot %d", b.Slot)) + } if onGraphBuilt != nil { onGraphBuilt() } + dispatchDependencyPlan(plan, out, done) +} +// dispatchDependencyPlan consumes a one-shot prebuilt plan. It mutates only +// the plan's private in-degree counters while transaction workers report +// completion through done. +func dispatchDependencyPlan(plan *dependencyPlan, out chan int, done chan int) { sent := 0 - // Output a topological sorting of the transactions - for t, deg := range inDegree { - if deg == 0 { - out <- t + for transaction, degree := range plan.inDegree { + if degree == 0 { + out <- transaction sent++ } } - for sent < len(b.Transactions) { - in := <-done - for _, dependentTx := range adjList[int(in)] { - inDegree[int(dependentTx)]-- - if inDegree[int(dependentTx)] == 0 { + for sent < len(plan.inDegree) { + completed := <-done + start, end := plan.offsets[completed], plan.offsets[completed+1] + for _, dependentTx := range plan.dependents[start:end] { + plan.inDegree[dependentTx]-- + if plan.inDegree[dependentTx] == 0 { out <- int(dependentTx) sent++ } diff --git a/pkg/replay/topsort_planner_test.go b/pkg/replay/topsort_planner_test.go index 53edf4f43..066f9c6ac 100644 --- a/pkg/replay/topsort_planner_test.go +++ b/pkg/replay/topsort_planner_test.go @@ -1,6 +1,7 @@ package replay import ( + "encoding/binary" "encoding/json" "slices" "sort" @@ -226,6 +227,187 @@ func TestTopsort(t *testing.T) { } } +func plannerAccountsConflict(a, b plannerTransactionAccounts) bool { + bAccounts := make(map[solana.PublicKey]struct{}, len(b.accounts)) + for _, account := range b.accounts { + bAccounts[account.publicKey] = struct{}{} + } + for _, account := range a.writable() { + if _, exists := bAccounts[account.publicKey]; exists { + return true + } + } + + aAccounts := make(map[solana.PublicKey]struct{}, len(a.accounts)) + for _, account := range a.accounts { + aAccounts[account.publicKey] = struct{}{} + } + for _, account := range b.writable() { + if _, exists := aAccounts[account.publicKey]; exists { + return true + } + } + return false +} + +func plannerAccountModes(accounts plannerTransactionAccounts) map[solana.PublicKey]bool { + modes := make(map[solana.PublicKey]bool, len(accounts.accounts)) + for _, account := range accounts.readonly() { + modes[account.publicKey] = false + } + for _, account := range accounts.writable() { + modes[account.publicKey] = true + } + return modes +} + +func dependencyPlanReachableFrom(plan *dependencyPlan, from int) []bool { + visited := make([]bool, len(plan.inDegree)) + stack := []uint32{uint32(from)} + visited[from] = true + for len(stack) > 0 { + current := stack[len(stack)-1] + stack = stack[:len(stack)-1] + start, end := plan.offsets[current], plan.offsets[current+1] + for _, dependent := range plan.dependents[start:end] { + if !visited[dependent] { + visited[dependent] = true + stack = append(stack, dependent) + } + } + } + return visited +} + +func assertConflictOrderReachable(t *testing.T, accounts []plannerTransactionAccounts, plan *dependencyPlan) { + t.Helper() + for earlier := range accounts { + reachable := dependencyPlanReachableFrom(plan, earlier) + for later := earlier + 1; later < len(accounts); later++ { + if !plannerAccountsConflict(accounts[earlier], accounts[later]) { + continue + } + if !reachable[later] { + t.Fatalf("conflicting transaction %d does not reach later transaction %d", earlier, later) + } + } + } +} + +func TestDependencyPlanPreservesConflictOrder(t *testing.T) { + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + accounts, available := plannerAccountsForBlock(test.b) + if !available { + t.Fatal("dependency planner unavailable") + } + assertConflictOrderReachable(t, accounts, buildDependencyPlan(accounts)) + }) + } +} + +func TestDependencyPlanPrunesTransitiveWriterEdges(t *testing.T) { + const transactionCount = 1_000 + sharedAccount := testPk(0x7f) + transactions := make([]plannerTransactionAccounts, transactionCount) + for idx := range transactions { + transactions[idx] = newPlannerTransactionAccounts(1, func(add func(solana.PublicKey, bool)) { + add(sharedAccount, true) + }) + } + + plan := buildDependencyPlan(transactions) + if got, want := len(plan.dependents), transactionCount-1; got != want { + t.Fatalf("dependency edge count = %d, want chain of %d", got, want) + } + for idx := 0; idx < transactionCount-1; idx++ { + start, end := plan.offsets[idx], plan.offsets[idx+1] + if end-start != 1 || plan.dependents[start] != uint32(idx+1) { + t.Fatalf("transaction %d dependents = %v, want [%d]", idx, plan.dependents[start:end], idx+1) + } + } +} + +func TestPlannerAccountDeduplicationPromotesWritable(t *testing.T) { + account := testPk(0x5a) + accounts := newPlannerTransactionAccounts(2, func(add func(solana.PublicKey, bool)) { + add(account, false) + add(account, true) + }) + if len(accounts.readonly()) != 0 { + t.Fatalf("readonly accounts = %v, want none", accounts.readonly()) + } + writable := make([]solana.PublicKey, len(accounts.writable())) + for idx, access := range accounts.writable() { + writable[idx] = access.publicKey + } + if diff := cmp.Diff([]solana.PublicKey{account}, writable); diff != "" { + t.Fatalf("writable accounts mismatch (-want +got):\n%s", diff) + } +} + +func TestPlannerResolvedMessageAndMetadataAgree(t *testing.T) { + payer := testPk(0x61) + table := testPk(0x62) + loadedWritable := testPk(0x63) + loadedReadonly := testPk(0x64) + msg := solana.Message{ + Header: solana.MessageHeader{ + NumRequiredSignatures: 1, + }, + AccountKeys: []solana.PublicKey{payer}, + AddressTableLookups: solana.MessageAddressTableLookupSlice{ + { + AccountKey: table, + WritableIndexes: []uint8{0}, + ReadonlyIndexes: []uint8{1}, + }, + }, + } + msg.SetVersion(solana.MessageVersionV0) + transaction := &solana.Transaction{Message: msg} + txMeta := &rpc.TransactionMeta{ + LoadedAddresses: rpc.LoadedAddresses{ + Writable: []solana.PublicKey{loadedWritable}, + ReadOnly: []solana.PublicKey{loadedReadonly}, + }, + } + beforeResolution := plannerAccountsFromMeta(transaction, txMeta) + + if err := transaction.Message.SetAddressTables(map[solana.PublicKey]solana.PublicKeySlice{ + table: {loadedWritable, loadedReadonly}, + }); err != nil { + t.Fatalf("set address tables: %v", err) + } + if err := transaction.Message.ResolveLookups(); err != nil { + t.Fatalf("resolve lookups: %v", err) + } + + afterResolutionFromMeta := plannerAccountsFromMeta(transaction, txMeta) + afterResolutionFromMessage := plannerAccountsFromMessage(&transaction.Message) + if diff := cmp.Diff(plannerAccountModes(beforeResolution), plannerAccountModes(afterResolutionFromMeta)); diff != "" { + t.Fatalf("metadata access changed across resolution (-before +after):\n%s", diff) + } + if diff := cmp.Diff(plannerAccountModes(beforeResolution), plannerAccountModes(afterResolutionFromMessage)); diff != "" { + t.Fatalf("resolved-message access differs from metadata (-meta +message):\n%s", diff) + } + + resolvedBlock := &block.Block{ + Transactions: []*solana.Transaction{transaction}, + TxMetas: []*rpc.TransactionMeta{{}}, + } + blockAccounts, available := plannerAccountsForBlock(resolvedBlock) + if !available { + t.Fatal("planner unavailable for resolved message with incomplete metadata") + } + if diff := cmp.Diff( + plannerAccountModes(afterResolutionFromMessage), + plannerAccountModes(blockAccounts[0]), + ); diff != "" { + t.Fatalf("incomplete metadata hid resolved lookup accounts (-message +block):\n%s", diff) + } +} + func mustMarshal(b *block.Block) []byte { bBytes, err := json.Marshal(b) if err != nil { @@ -262,6 +444,11 @@ func FuzzBlockToDependencyGraph(f *testing.F) { } } + plannerAccounts, available := plannerAccountsForBlock(b) + if !available { + t.Skip("skipping block whose planner accounts are unavailable") + } + plan := buildDependencyPlan(plannerAccounts) adjList, inDegrees := blockToDependencyGraph(b) if len(adjList) != len(b.Transactions) { t.Errorf("len(adjList)=%d != len(b.Transactions)=%d", len(adjList), len(b.Transactions)) @@ -291,5 +478,104 @@ func FuzzBlockToDependencyGraph(f *testing.F) { t.Errorf("node=%d neighbors list=%+v had duplicates %v", u, uncompactedVs, slices.Compact(vs0)) } } + if len(plannerAccounts) <= 128 { + assertConflictOrderReachable(t, plannerAccounts, plan) + } + }) +} + +func FuzzDependencyPlanPreservesConflictOrder(f *testing.F) { + f.Add([]byte{4, 1, 0, 1, 1, 2, 0, 2, 1}) + f.Add([]byte{8, 3, 1, 2, 0, 4, 1, 3, 0, 2, 1}) + f.Fuzz(func(t *testing.T, data []byte) { + if len(data) < 2 { + t.Skip() + } + transactionCount := 1 + int(data[0]%64) + transactions := make([]plannerTransactionAccounts, transactionCount) + cursor := 1 + nextByte := func() byte { + value := data[cursor%len(data)] + cursor++ + return value + } + for idx := range transactions { + accessCount := 1 + int(nextByte()%8) + builder := newPlannerTransactionAccountBuilder(accessCount) + for range accessCount { + account := testPk(nextByte() % 32) + writable := nextByte()&1 != 0 + builder.add(account, writable) + } + transactions[idx] = builder.finish() + } + + plan := buildDependencyPlan(transactions) + for prerequisite := range plan.inDegree { + start, end := plan.offsets[prerequisite], plan.offsets[prerequisite+1] + for _, dependent := range plan.dependents[start:end] { + if int(dependent) <= prerequisite { + t.Fatalf("edge %d -> %d does not point forward", prerequisite, dependent) + } + } + } + assertConflictOrderReachable(t, transactions, plan) }) } + +var benchmarkDependencyPlanSink *dependencyPlan + +func benchmarkPlannerBlock(transactionCount int, sharedWriter bool) *block.Block { + transactions := make([]*solana.Transaction, transactionCount) + sharedAccount := testPk(0xe0) + readonlyProgram := testPk(0xe1) + for idx := range transactions { + payer := solana.PublicKey{} + binary.LittleEndian.PutUint64(payer[:8], uint64(idx+1)) + if sharedWriter { + payer = sharedAccount + } + uniqueWritable := solana.PublicKey{} + binary.LittleEndian.PutUint64(uniqueWritable[:8], uint64(transactionCount+idx+1)) + uniqueReadonly := solana.PublicKey{} + binary.LittleEndian.PutUint64(uniqueReadonly[:8], uint64(2*transactionCount+idx+1)) + transactions[idx] = &solana.Transaction{ + Message: solana.Message{ + Header: solana.MessageHeader{ + NumRequiredSignatures: 1, + NumReadonlyUnsignedAccounts: 2, + }, + AccountKeys: []solana.PublicKey{ + payer, + uniqueWritable, + readonlyProgram, + uniqueReadonly, + }, + }, + } + } + return &block.Block{Transactions: transactions} +} + +func BenchmarkDependencyPlan(b *testing.B) { + for _, test := range []struct { + name string + sharedWriter bool + }{ + {name: "independent", sharedWriter: false}, + {name: "shared_writer", sharedWriter: true}, + } { + b.Run(test.name, func(b *testing.B) { + block := benchmarkPlannerBlock(10_000, test.sharedWriter) + b.ReportAllocs() + b.ResetTimer() + for range b.N { + plan, available := dependencyPlanForBlock(block) + if !available { + b.Fatal("dependency planner unavailable") + } + benchmarkDependencyPlanSink = plan + } + }) + } +} diff --git a/pkg/replay/transaction.go b/pkg/replay/transaction.go index 0d15b2c62..bd2ffffa4 100644 --- a/pkg/replay/transaction.go +++ b/pkg/replay/transaction.go @@ -512,24 +512,6 @@ func verifySignatures(snapshot *sigverifySnapshot, sigverifyWg *sync.WaitGroup) metrics.GlobalBlockReplay.Sigverify.AddTimingSince(start) } -func cloneTransaction(tx *solana.Transaction) (*solana.Transaction, error) { - if tx == nil { - return nil, nil - } - - raw, err := tx.MarshalBinary() - if err != nil { - return nil, err - } - - cloned, err := solana.TransactionFromBytes(raw) - if err != nil { - return nil, err - } - - return cloned, nil -} - func processTransactionComputeUnits(execCtx *sealevel.ExecutionCtx) uint64 { if execCtx == nil { return 0 diff --git a/scripts/replay_timings_viewer.py b/scripts/replay_timings_viewer.py index a24345ea9..943795ad5 100644 --- a/scripts/replay_timings_viewer.py +++ b/scripts/replay_timings_viewer.py @@ -100,8 +100,9 @@ def _(alt, latency_records, mo): @app.cell(hide_code=True) def _(alt, latency_records, mo): - # Only disjoint top-level ProcessBlock phases are stacked. Planner build - # and dispatch are nested within TxLoop, so they are overlaid as lines. + # Only disjoint top-level ProcessBlock phases are stacked. Prepared planner + # build may overlap account loading; wait and dispatch are nested within + # TxLoop, so all three are overlaid as diagnostic lines. # SignatureVerificationJoin is only the final blocking wait; the existing # Sigverify metric is summed worker time that overlaps these wall phases. process_components = ( @@ -149,7 +150,11 @@ def _(alt, latency_records, mo): planner_detail = ( alt.Chart(alt.InlineData(values=latency_records)) .transform_fold( - ["DependencyPlannerBuild", "DependencyPlannerDispatch"], + [ + "DependencyPlannerBuild", + "DependencyPlannerWait", + "DependencyPlannerDispatch", + ], as_=["Nested timer", "Latency"], ) .mark_line(point=True, strokeDash=[5, 3]) @@ -164,7 +169,7 @@ def _(alt, latency_records, mo): alt.layer(process_components, process_total, planner_detail) .resolve_scale(color="independent") .properties( - title="ProcessBlock detail (black total; dashed planner timers are nested in TxLoop)" + title="ProcessBlock detail (black total; prepared build may overlap load, wait/dispatch are nested in TxLoop)" ) ) mo.ui.altair_chart(process_chart)