-
Notifications
You must be signed in to change notification settings - Fork 468
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
11 changed files
with
305 additions
and
29 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,64 @@ | ||
| package server_api | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "time" | ||
|
|
||
| "github.com/ethereum/go-ethereum/common" | ||
| "github.com/ethereum/go-ethereum/log" | ||
| "github.com/offchainlabs/nitro/pubsub" | ||
| "github.com/offchainlabs/nitro/util/stopwaiter" | ||
| "github.com/offchainlabs/nitro/validator" | ||
| "github.com/offchainlabs/nitro/validator/server_api/validation" | ||
| ) | ||
|
|
||
| // RedisValidationServer implements consumer for the requests originated from | ||
| // RedisValidationClient producers. | ||
| type RedisValidationServer struct { | ||
| stopwaiter.StopWaiter | ||
| spawner validator.ValidationSpawner | ||
|
|
||
| // consumers stores moduleRoot to consumer mapping. | ||
| consumers map[common.Hash]*pubsub.Consumer[*validator.ValidationInput, validator.GoGlobalState] | ||
| } | ||
|
|
||
| func NewRedisValidationServer(cfg *validation.RedisValidationServerConfig) (*RedisValidationServer, error) { | ||
| res := &RedisValidationServer{} | ||
| for _, mr := range cfg.ModuleRoots { | ||
| conf := cfg.ConsumerConfig.Clone() | ||
| conf.RedisStream, conf.RedisGroup = redisStreamForRoot(mr), redisGroupForRoot(mr) | ||
| c, err := pubsub.NewConsumer[*validator.ValidationInput, validator.GoGlobalState](&conf) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("creating consumer for validation: %w", err) | ||
| } | ||
| res.consumers[mr] = c | ||
| } | ||
| return res, nil | ||
| } | ||
|
|
||
| func (s *RedisValidationServer) Start(ctx_in context.Context) { | ||
| s.StopWaiter.Start(ctx_in, s) | ||
| for moduleRoot, c := range s.consumers { | ||
| c := c | ||
| c.Start(ctx_in) | ||
| s.StopWaiter.CallIteratively(func(ctx context.Context) time.Duration { | ||
| req, err := c.Consume(ctx) | ||
| if err != nil { | ||
| log.Error("Consuming request", "error", err) | ||
| return 0 | ||
| } | ||
| valRun := s.spawner.Launch(req.Value, moduleRoot) | ||
| res, err := valRun.Await(ctx) | ||
| if err != nil { | ||
| log.Error("Error validating", "input", "request value", req.Value, "error", err) | ||
| return 0 | ||
| } | ||
| if err := c.SetResult(ctx, req.ID, res); err != nil { | ||
| log.Error("Error setting result for request", "id", req.ID, "result", res, "error", err) | ||
| return 0 | ||
| } | ||
| return time.Second | ||
| }) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,116 @@ | ||
| package server_api | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
|
|
||
| "github.com/ethereum/go-ethereum/common" | ||
| "github.com/offchainlabs/nitro/pubsub" | ||
| "github.com/offchainlabs/nitro/util/containers" | ||
| "github.com/offchainlabs/nitro/util/stopwaiter" | ||
| "github.com/offchainlabs/nitro/validator" | ||
| "github.com/offchainlabs/nitro/validator/server_common" | ||
| "github.com/spf13/pflag" | ||
| ) | ||
|
|
||
| type RedisValidationClientConfig struct { | ||
| Name string `koanf:"name"` | ||
| Room int32 `koanf:"room"` | ||
| ProducerConfig pubsub.ProducerConfig `koanf:"producer-config"` | ||
| // Supported wasm module roots. | ||
| ModuleRoots []common.Hash `koanf:"module-roots"` | ||
| } | ||
|
|
||
| var DefaultRedisValidationClientConfig = &RedisValidationClientConfig{ | ||
| Name: "redis validation client", | ||
| Room: 2, | ||
| ProducerConfig: pubsub.DefaultProducerConfig, | ||
| } | ||
|
|
||
| var TestRedisValidationClientConfig = &RedisValidationClientConfig{ | ||
| Name: "test redis validation client", | ||
| Room: 2, | ||
| ProducerConfig: pubsub.TestProducerConfig, | ||
| } | ||
|
|
||
| func RedisValidationClientConfigAddOptions(prefix string, f *pflag.FlagSet) { | ||
| f.String(prefix+".name", DefaultRedisValidationClientConfig.Name, "validation client name") | ||
| f.Uint64(prefix+".room", uint64(DefaultRedisValidationClientConfig.Room), "validation client room") | ||
| pubsub.ProducerAddConfigAddOptions(prefix+".producer-config", f) | ||
| // TODO(anodar): initialize module roots here. | ||
| } | ||
|
|
||
| // RedisValidationClient implements validation client through redis streams. | ||
| type RedisValidationClient struct { | ||
| stopwaiter.StopWaiter | ||
| name string | ||
| room int32 | ||
| // producers stores moduleRoot to producer mapping. | ||
| producers map[common.Hash]*pubsub.Producer[*validator.ValidationInput, validator.GoGlobalState] | ||
| } | ||
|
|
||
| func redisGroupForRoot(moduleRoot common.Hash) string { | ||
| return fmt.Sprintf("group:%s", moduleRoot.Hex()) | ||
| } | ||
|
|
||
| func redisStreamForRoot(moduleRoot common.Hash) string { | ||
| return fmt.Sprintf("group:%s", moduleRoot.Hex()) | ||
| } | ||
|
|
||
| func NewRedisValidationClient(cfg *RedisValidationClientConfig) (*RedisValidationClient, error) { | ||
| res := &RedisValidationClient{ | ||
| name: cfg.Name, | ||
| room: cfg.Room, | ||
| producers: make(map[common.Hash]*pubsub.Producer[*validator.ValidationInput, validator.GoGlobalState]), | ||
| } | ||
| for _, mr := range cfg.ModuleRoots { | ||
| c := cfg.ProducerConfig.Clone() | ||
| c.RedisStream, c.RedisGroup = redisGroupForRoot(mr), redisStreamForRoot(mr) | ||
| p, err := pubsub.NewProducer[*validator.ValidationInput, validator.GoGlobalState](&c) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("creating producer for validation: %w", err) | ||
| } | ||
| res.producers[mr] = p | ||
| } | ||
| return res, nil | ||
| } | ||
|
|
||
| func (c *RedisValidationClient) Launch(entry *validator.ValidationInput, moduleRoot common.Hash) validator.ValidationRun { | ||
| producer, found := c.producers[moduleRoot] | ||
| if !found { | ||
| errPromise := containers.NewReadyPromise(validator.GoGlobalState{}, fmt.Errorf("no validation is configured for wasm root %v", moduleRoot)) | ||
| return server_common.NewValRun(errPromise, moduleRoot) | ||
| } | ||
| promise, err := producer.Produce(c.GetContext(), entry) | ||
| if err != nil { | ||
| errPromise := containers.NewReadyPromise(validator.GoGlobalState{}, fmt.Errorf("error producing input: %w", err)) | ||
| return server_common.NewValRun(errPromise, moduleRoot) | ||
| } | ||
| return server_common.NewValRun(promise, moduleRoot) | ||
| } | ||
|
|
||
| func (c *RedisValidationClient) Start(ctx_in context.Context) error { | ||
| for _, p := range c.producers { | ||
| p.Start(ctx_in) | ||
| } | ||
| c.StopWaiter.Start(ctx_in, c) | ||
| return nil | ||
| } | ||
|
|
||
| func (c *RedisValidationClient) Stop() { | ||
| for _, p := range c.producers { | ||
| p.StopAndWait() | ||
| } | ||
| c.StopWaiter.StopAndWait() | ||
| } | ||
|
|
||
| func (c *RedisValidationClient) Name() string { | ||
| if c.Started() { | ||
| return c.name | ||
| } | ||
| return "(not started)" | ||
| } | ||
|
|
||
| func (c *RedisValidationClient) Room() int { | ||
| return int(c.room) | ||
| } |
Oops, something went wrong.