Process messages concurrently #288
Conversation
This reverts commit 3b0c438.
relayer/application_relayer.go
Outdated
| var eg errgroup.Group | ||
| for _, msg := range msgs { | ||
| eg.Go(func() error { | ||
| return r.ProcessMessage(msg) |
There was a problem hiding this comment.
Upgraded to go 1.22 so that this closure captures the value of msg on each iteration, rather than the variable itself.
There was a problem hiding this comment.
That's an interesting change for a language upgrade. I'm surprised if this change doesn't break some apps.
There was a problem hiding this comment.
@cam-schultz There's a flag you can set in 1.21 that enables this feature, so we can keep our versions consistent.
There was a problem hiding this comment.
There was a problem hiding this comment.
Thanks for pointing that out @geoff-vball . I elected to go with the workaround of assigning a local variable in each iteration rather than include that experimental flag in the build, since we're trying to lock into stable language features.
Refactor application relayers
geoff-vball
dismissed stale reviews from michaelkaplan13 and themself
via
7d2f8ee
|
|
||
| // Check for the expected number of responses, and clear from the map if all expected responses have been received | ||
| // TODO: we can improve performance here by independently locking the response channel and response count maps | ||
| responses := h.responsesCount[requestID] |
There was a problem hiding this comment.
nit: perhaps add an "ok" check on the map lookup just to defensively check for logic errors.
Why this should be merged
Fixes #31
Refactors
ApplicationRelayerto be more akin to a standalone worker, as opposed to a composite piece of a subnetListenerHow this works
ProcessHeightandProcessMessagetoApplicationRelayer.ProcessHeightcallsProcessMessagein a separete goroutine for each Warp message in the block.ProcessMessagecan be called on its own (as is the case for manual Warp message relays)ApplicationRelayerfromListener,ExternalHandler, andMessageManagerExternalHandleradds the methodRegisterRequestID, which routes all inbound app responses for that ID to the caller. Used to route messages directly to the initiatingApplicationRelayerMessageManageris now a factory forMessageHandlers, an ephemeral type that handles message processing for a single Warp messageCheckpointManagersimplified to only perform periodic writes. The previous prepare-stage-commit flow is now encapsulated inApplicationRelayer.ProcessHeightBatchMessagee2e test that sends multiple Teleporter messages in a single transaction. This tests concurrent message processing within a single block. Solidity infra added totests/to support the newBatchCrossChainMessagecontract.How this was tested
CI
Added batch tx e2e test
TODO: before/after througput testing (see #185)
How is this documented
N/A. No functional changes, pure throughput