Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: Optimistic Execution #16581

Merged
merged 39 commits into from
Sep 18, 2023
Merged
Show file tree
Hide file tree
Changes from 14 commits
Commits
Show all changes
39 commits
Select commit Hold shift + click to select a range
a57c937
feat: Optimistic Execution
facundomedica Jun 15, 2023
023256e
fix panic recovery
facundomedica Jun 15, 2023
14b80c4
remove test changes
facundomedica Jun 15, 2023
47b8a1c
Merge branch 'main' of https://github.com/cosmos/cosmos-sdk into facu/oe
facundomedica Jun 16, 2023
deaf6b7
fix test
facundomedica Jun 16, 2023
f30e4a7
make comet panic instead of sdk
facundomedica Jun 16, 2023
573d107
add abort channel
facundomedica Jun 20, 2023
17b5ca4
fix abort
facundomedica Jun 20, 2023
d371c16
clean up phase1
facundomedica Jun 20, 2023
c9dbc9a
merge
facundomedica Jun 30, 2023
20f0325
testing testing
facundomedica Jun 30, 2023
6aec99a
Merge branch 'main' of https://github.com/cosmos/cosmos-sdk into facu/oe
facundomedica Jul 13, 2023
e920201
merge main
facundomedica Jul 20, 2023
b855c1a
progress
facundomedica Jul 20, 2023
265e32d
fix
facundomedica Jul 21, 2023
2830366
Merge branch 'main' of https://github.com/cosmos/cosmos-sdk into facu/oe
facundomedica Jul 26, 2023
c835fa7
progress
facundomedica Jul 27, 2023
b26cfe8
Merge branch 'main' into facu/oe
facundomedica Jul 27, 2023
06cb990
lint
facundomedica Jul 27, 2023
f2aec1d
progress
facundomedica Jul 27, 2023
64988fa
Merge branch 'main' of https://github.com/cosmos/cosmos-sdk into facu/oe
facundomedica Jul 31, 2023
18b666e
fix race condition
facundomedica Jul 31, 2023
125e942
progress
facundomedica Jul 31, 2023
0f1ad3b
progress
facundomedica Aug 1, 2023
35ae374
Merge branch 'main' of https://github.com/cosmos/cosmos-sdk into facu/oe
facundomedica Aug 2, 2023
c798e17
progress
facundomedica Aug 2, 2023
655dde4
merge main
facundomedica Aug 17, 2023
74147f1
added mutext to mempools
facundomedica Aug 17, 2023
0d45c3c
add test and do some refactor
facundomedica Aug 27, 2023
b008a8a
undo test changes
facundomedica Aug 27, 2023
78b233d
fix
facundomedica Aug 27, 2023
4f90f04
Update baseapp/abci.go
facundomedica Aug 29, 2023
2b574d5
only start optimistic execution if processProposal resp is accepted
facundomedica Sep 9, 2023
f91b715
Merge branch 'main' into facu/oe
facundomedica Sep 14, 2023
1c4743a
godoc + tests
facundomedica Sep 18, 2023
8cda1f1
add file
facundomedica Sep 18, 2023
0065196
cl++
facundomedica Sep 18, 2023
9d6c8b1
cl++
facundomedica Sep 18, 2023
8bdd23d
lint
facundomedica Sep 18, 2023
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
78 changes: 65 additions & 13 deletions baseapp/abci.go
Original file line number Diff line number Diff line change
Expand Up @@ -492,6 +492,12 @@
// processed the first block, as we want to avoid overwriting the finalizeState
// after state changes during InitChain.
if req.Height > app.initialHeight {
// abort any running OE
if app.oeEnabled && app.oeInfo != nil && app.oeInfo.Running() {
app.oeInfo.Abort()
_, _ = app.oeInfo.WaitResult() // ignore the result
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Change potentially affects state.

Call sequence:

(*github.com/cosmos/cosmos-sdk/baseapp.BaseApp).ProcessProposal (baseapp/abci.go:467)

}

app.setState(execModeFinalize, header)
}

Expand Down Expand Up @@ -532,6 +538,13 @@
return &abci.ResponseProcessProposal{Status: abci.ResponseProcessProposal_REJECT}, nil
}

// Only execute optimistic execution if OE is enabled and the block height is greater than the initial height.
// During the first block we'll be carrying state from InitChain, so it would be impossible for us to easily revert.
facundomedica marked this conversation as resolved.
Show resolved Hide resolved
if app.oeEnabled && req.Height > app.initialHeight {
app.oeInfo = SetupOptimisticExecution(req, app.internalFinalizeBlock)
app.oeInfo.Execute()
}

return resp, nil
}

Expand Down Expand Up @@ -640,17 +653,7 @@
return resp, err
}

// FinalizeBlock will execute the block proposal provided by RequestFinalizeBlock.
// Specifically, it will execute an application's BeginBlock (if defined), followed
// by the transactions in the proposal, finally followed by the application's
// EndBlock (if defined).
//
// For each raw transaction, i.e. a byte slice, BaseApp will only execute it if
// it adheres to the sdk.Tx interface. Otherwise, the raw transaction will be
// skipped. This is to support compatibility with proposers injecting vote
// extensions into the proposal, which should not themselves be executed in cases
// where they adhere to the sdk.Tx interface.
func (app *BaseApp) FinalizeBlock(req *abci.RequestFinalizeBlock) (*abci.ResponseFinalizeBlock, error) {
func (app *BaseApp) internalFinalizeBlock(req *abci.RequestFinalizeBlock) (*abci.ResponseFinalizeBlock, error) {
var events []abci.Event

if err := app.checkHalt(req.Height, req.Time); err != nil {
Expand Down Expand Up @@ -720,7 +723,17 @@
}
}

beginBlock := app.beginBlock(req)
beginBlock, err := app.beginBlock(req)
if err != nil {
return nil, err
}

// First check for an abort signal after beginBlock, as it's the first place
// we spend any significant amount of time.
if app.oeInfo != nil && app.oeInfo.ShouldAbort() {
return nil, nil
}

events = append(events, beginBlock.Events...)

// Iterate over all raw transactions in the proposal and attempt to execute
Expand All @@ -730,6 +743,11 @@
// vote extensions, so skip those.
txResults := make([]*abci.ExecTxResult, 0, len(req.Txs))
for _, rawTx := range req.Txs {
// check before every tx if we should abort
if app.oeInfo != nil && app.oeInfo.ShouldAbort() {
return nil, nil
}

var response *abci.ExecTxResult

if _, err := app.txDecoder(rawTx); err == nil {
Expand Down Expand Up @@ -767,10 +785,44 @@
TxResults: txResults,
ValidatorUpdates: endBlock.ValidatorUpdates,
ConsensusParamUpdates: &cp,
AppHash: app.workingHash(),
// AppHash: app.workingHash(),
}, nil
}

// FinalizeBlock will execute the block proposal provided by RequestFinalizeBlock.
// Specifically, it will execute an application's BeginBlock (if defined), followed
// by the transactions in the proposal, finally followed by the application's
// EndBlock (if defined).
//
// For each raw transaction, i.e. a byte slice, BaseApp will only execute it if
// it adheres to the sdk.Tx interface. Otherwise, the raw transaction will be
// skipped. This is to support compatibility with proposers injecting vote
// extensions into the proposal, which should not themselves be executed in cases
// where they adhere to the sdk.Tx interface.
func (app *BaseApp) FinalizeBlock(req *abci.RequestFinalizeBlock) (*abci.ResponseFinalizeBlock, error) {
if app.oeInfo != nil && app.oeEnabled {
// check if the hash we got is the same as the one we are executing
aborted := app.oeInfo.AbortIfNeeded(req.Hash)
// Wait for the OE to finish, regardless of whether it was aborted or not
res, err := app.oeInfo.WaitResult()

// only return if we are not aborting
if !aborted {
res.AppHash = app.workingHash()
return res, err
}

// if it was aborted, we need to reset the state and continue
app.finalizeBlockState = nil
}

// if no OE is running, just run the block (this is either a block replay or a OE that got aborted)
app.oeInfo = nil
res, err := app.internalFinalizeBlock(req)
res.AppHash = app.workingHash()
Fixed Show fixed Hide fixed
return res, err
}

// checkHalt checkes if height or time exceeds halt-height or halt-time respectively.
func (app *BaseApp) checkHalt(height int64, time time.Time) error {
var halt bool
Expand Down
12 changes: 8 additions & 4 deletions baseapp/baseapp.go
Original file line number Diff line number Diff line change
Expand Up @@ -180,6 +180,9 @@ type BaseApp struct {
chainID string

cdc codec.Codec

oeInfo *OptimisticExecutionInfo
oeEnabled bool
}

// NewBaseApp returns a reference to an initialized BaseApp. It accepts a
Expand All @@ -198,6 +201,7 @@ func NewBaseApp(
msgServiceRouter: NewMsgServiceRouter(),
txDecoder: txDecoder,
fauxMerkleMode: false,
oeEnabled: true,
}

for _, option := range options {
Expand Down Expand Up @@ -668,7 +672,7 @@ func (app *BaseApp) cacheTxContext(ctx sdk.Context, txBytes []byte) (sdk.Context
return ctx.WithMultiStore(msCache), msCache
}

func (app *BaseApp) beginBlock(req *abci.RequestFinalizeBlock) sdk.BeginBlock {
func (app *BaseApp) beginBlock(req *abci.RequestFinalizeBlock) (sdk.BeginBlock, error) {
var (
resp sdk.BeginBlock
err error
Expand All @@ -677,7 +681,7 @@ func (app *BaseApp) beginBlock(req *abci.RequestFinalizeBlock) sdk.BeginBlock {
if app.beginBlocker != nil {
resp, err = app.beginBlocker(app.finalizeBlockState.ctx)
if err != nil {
panic(err)
return resp, err
}

// append BeginBlock attributes to all events in the EndBlock response
Expand All @@ -691,7 +695,7 @@ func (app *BaseApp) beginBlock(req *abci.RequestFinalizeBlock) sdk.BeginBlock {
resp.Events = sdk.MarkEventsToIndex(resp.Events, app.indexEvents)
}

return resp
return resp, nil
}

func (app *BaseApp) deliverTx(tx []byte) *abci.ExecTxResult {
Expand Down Expand Up @@ -739,7 +743,7 @@ func (app *BaseApp) endBlock(ctx context.Context) (sdk.EndBlock, error) {
if app.endBlocker != nil {
eb, err := app.endBlocker(app.finalizeBlockState.ctx)
if err != nil {
panic(err)
return endblock, err
}

// append EndBlock attributes to all events in the EndBlock response
Expand Down
100 changes: 100 additions & 0 deletions baseapp/optimistic_execution.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,100 @@
package baseapp

import (
"bytes"
"log"
"math/rand"
"sync"
"time"

abci "github.com/cometbft/cometbft/abci/types"
)

type OptimisticExecutionInfo struct {
mtx sync.RWMutex
stopCh chan struct{}
shouldAbort bool
running bool

// we could use generics here in the future to allow other types of req/resp
fn func(*abci.RequestFinalizeBlock) (*abci.ResponseFinalizeBlock, error)
Request *abci.RequestFinalizeBlock
Response *abci.ResponseFinalizeBlock
Error error
executionTime time.Duration
}

func SetupOptimisticExecution(
req *abci.RequestProcessProposal,
fn func(*abci.RequestFinalizeBlock) (*abci.ResponseFinalizeBlock, error),
) *OptimisticExecutionInfo {
return &OptimisticExecutionInfo{
stopCh: make(chan struct{}),
fn: fn,
Request: &abci.RequestFinalizeBlock{
Txs: req.Txs,
DecidedLastCommit: req.ProposedLastCommit,
Misbehavior: req.Misbehavior,
Hash: req.Hash,
Height: req.Height,
Time: req.Time,
NextValidatorsHash: req.NextValidatorsHash,
ProposerAddress: req.ProposerAddress,
},
}
}

func (oe *OptimisticExecutionInfo) Execute() {
log.Println("Start OE ✅")
start := time.Now()
Fixed Show fixed Hide fixed
oe.running = true
go func() {
resp, err := oe.fn(oe.Request)
oe.mtx.Lock()
oe.executionTime = time.Since(start)
oe.Response, oe.Error = resp, err
oe.running = false
close(oe.stopCh)
oe.mtx.Unlock()
}()
Fixed Show fixed Hide fixed
}

// AbortIfNeeded
// If the request hash is not the same as the one in the OE, then abort the OE
// and wait for the abort to happen. Returns true if the OE was aborted.
func (oe *OptimisticExecutionInfo) AbortIfNeeded(reqHash []byte) bool {
oe.mtx.Lock()
defer oe.mtx.Unlock()
if rand.Intn(100) > 80 || !bytes.Equal(oe.Request.Hash, reqHash) {
Fixed Show fixed Hide fixed
log.Println("OE aborted ❌")
oe.shouldAbort = true
}
return oe.shouldAbort
}

func (oe *OptimisticExecutionInfo) Abort() {
oe.mtx.Lock()
defer oe.mtx.Unlock()
oe.shouldAbort = true
}

// ShouldAbort must only be used in the fn passed to SetupOptimisticExecution to
// check if the OE was aborted and return as soon as possible.
// TODO: figure out a better name, maybe ReturnEarly?
func (oe *OptimisticExecutionInfo) ShouldAbort() bool {
defer oe.mtx.RUnlock()
oe.mtx.RLock()
return oe.shouldAbort
}

func (oe *OptimisticExecutionInfo) Running() bool {
defer oe.mtx.RUnlock()
oe.mtx.RLock()
return oe.running
}

func (oe *OptimisticExecutionInfo) WaitResult() (*abci.ResponseFinalizeBlock, error) {
<-oe.stopCh
log.Println("OE took ⏱", oe.executionTime)
return oe.Response, oe.Error
}
Loading