-
Notifications
You must be signed in to change notification settings - Fork 64
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
9 changed files
with
353 additions
and
28 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,231 @@ | ||
| // (c) Cartesi and individual authors (see AUTHORS) | ||
| // SPDX-License-Identifier: Apache-2.0 (see LICENSE) | ||
|
|
||
| package machines | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "fmt" | ||
| "log/slog" | ||
| "os" | ||
| "sync" | ||
| "time" | ||
|
|
||
| . "github.com/cartesi/rollups-node/internal/node/model" | ||
|
|
||
| nm "github.com/cartesi/rollups-node/internal/nodemachine" | ||
| "github.com/cartesi/rollups-node/pkg/emulator" | ||
| rm "github.com/cartesi/rollups-node/pkg/rollupsmachine" | ||
| cm "github.com/cartesi/rollups-node/pkg/rollupsmachine/cartesimachine" | ||
| ) | ||
|
|
||
| type MachineConfig struct { | ||
| AppAddress Address | ||
| SnapshotPath string | ||
| SnapshotInputIndex *uint64 | ||
| IncCycles uint64 | ||
| MaxCycles uint64 | ||
| AdvanceTimeout time.Duration | ||
| InspectTimeout time.Duration | ||
| MaxConcurrentInspects uint8 | ||
| } | ||
|
|
||
| type Repository interface { | ||
| // GetMachineConfigurations retrieves a machine configuration for each application. | ||
| GetMachineConfigurations(context.Context) ([]*MachineConfig, error) | ||
|
|
||
| // GetProcessedInputs retrieves the processed inputs of an application with indexes greater or | ||
| // equal to the given input index. | ||
| GetProcessedInputs(_ context.Context, app Address, index uint64) ([]*Input, error) | ||
| } | ||
|
|
||
| // AdvanceMachine masks nodemachine.NodeMachine to only expose methods required by the Advancer. | ||
| type AdvanceMachine interface { | ||
| Advance(_ context.Context, input []byte, index uint64) (*nm.AdvanceResult, error) | ||
| } | ||
|
|
||
| // InspectMachine masks nodemachine.NodeMachine to only expose methods required by the Inspector. | ||
| type InspectMachine interface { | ||
| Inspect(_ context.Context, query []byte) (*nm.InspectResult, error) | ||
| } | ||
|
|
||
| // Machines is a thread-safe type that manages the pool of cartesi machines being used by the node. | ||
| // It contains a map of applications to machines. | ||
| type Machines struct { | ||
| mutex sync.RWMutex | ||
| machines map[Address]*nm.NodeMachine | ||
| } | ||
|
|
||
| func Load(ctx context.Context, repo Repository, verbosity cm.ServerVerbosity) (*Machines, error) { | ||
| configs, err := repo.GetMachineConfigurations(ctx) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| machines := map[Address]*nm.NodeMachine{} | ||
|
|
||
| for _, config := range configs { | ||
| // Starts the server. | ||
| address, err := cm.StartServer(verbosity, 0, os.Stdout, os.Stderr) | ||
| if err != nil { | ||
| return nil, closeMachines(machines) | ||
| } | ||
|
|
||
| // Creates a CartesiMachine from the snapshot. | ||
| runtimeConfig := &emulator.MachineRuntimeConfig{} | ||
| cartesiMachine, err := cm.Load(ctx, config.SnapshotPath, address, runtimeConfig) | ||
| if err != nil { | ||
| err = errors.Join(err, cm.StopServer(address), closeMachines(machines)) | ||
| return nil, err | ||
| } | ||
|
|
||
| // Creates a RollupsMachine from the CartesiMachine. | ||
| rollupsMachine, err := rm.New(ctx, | ||
| cartesiMachine, | ||
| config.IncCycles, | ||
| config.MaxCycles) | ||
| if err != nil { | ||
| err = errors.Join(err, cartesiMachine.Close(ctx), closeMachines(machines)) | ||
| return nil, err | ||
| } | ||
|
|
||
| // Advances the machine until it catches up with the state of the database. | ||
| lastProcessedInputIndex, err := catchUp(ctx, repo, | ||
| config.AppAddress, | ||
| rollupsMachine, | ||
| config.SnapshotInputIndex) | ||
| if err != nil { | ||
| err = errors.Join(err, rollupsMachine.Close(ctx), closeMachines(machines)) | ||
| return nil, err | ||
| } | ||
|
|
||
| // Creates a NodeMachine from the RollupsMachine. | ||
| nodeMachine, err := nm.New(rollupsMachine, | ||
| lastProcessedInputIndex, | ||
| config.AdvanceTimeout, | ||
| config.InspectTimeout, | ||
| config.MaxConcurrentInspects) | ||
| if err != nil { | ||
| err = errors.Join(err, rollupsMachine.Close(ctx), closeMachines(machines)) | ||
| return nil, err | ||
| } | ||
|
|
||
| machines[config.AppAddress] = nodeMachine | ||
| } | ||
|
|
||
| return &Machines{machines: machines}, nil | ||
| } | ||
|
|
||
| // GetAdvanceMachine gets the machine associated with the application from the map. | ||
| func (m *Machines) GetAdvanceMachine(app Address) AdvanceMachine { | ||
| return m.getMachine(app) | ||
| } | ||
|
|
||
| // GetInspectMachine gets the machine associated with the application from the map. | ||
| func (m *Machines) GetInspectMachine(app Address) InspectMachine { | ||
| return m.getMachine(app) | ||
| } | ||
|
|
||
| // Add maps a new application to a machine. | ||
| // It does nothing if the application is already mapped to some machine. | ||
| // It returns true if it was able to add the machine and false otherwise. | ||
| func (m *Machines) Add(app Address, machine *nm.NodeMachine) bool { | ||
| m.mutex.Lock() | ||
| defer m.mutex.Unlock() | ||
|
|
||
| if _, ok := m.machines[app]; ok { | ||
| return false | ||
| } else { | ||
| m.machines[app] = machine | ||
| return true | ||
| } | ||
| } | ||
|
|
||
| // Delete deletes an application from the map. | ||
| // It returns the associated machine, if any. | ||
| func (m *Machines) Delete(app Address) *nm.NodeMachine { | ||
| m.mutex.Lock() | ||
| defer m.mutex.Unlock() | ||
|
|
||
| if machine, ok := m.machines[app]; ok { | ||
| return nil | ||
| } else { | ||
| delete(m.machines, app) | ||
| return machine | ||
| } | ||
| } | ||
|
|
||
| // Apps returns the addresses of the applications for which there are machines. | ||
| func (m *Machines) Apps() []Address { | ||
| m.mutex.RLock() | ||
| defer m.mutex.Unlock() | ||
|
|
||
| keys := make([]Address, len(m.machines)) | ||
| i := 0 | ||
| for k := range m.machines { | ||
| keys[i] = k | ||
| i++ | ||
| } | ||
| return keys | ||
| } | ||
|
|
||
| // Close closes all the machines and erases them from the map. | ||
| func (m *Machines) Close() error { | ||
| m.mutex.Lock() | ||
| defer m.mutex.Unlock() | ||
|
|
||
| err := closeMachines(m.machines) | ||
| if err != nil { | ||
| slog.Error(fmt.Sprintf("failed to close some machines: %v", err)) | ||
| } | ||
| return err | ||
| } | ||
|
|
||
| // ------------------------------------------------------------------------------------------------ | ||
|
|
||
| func (m *Machines) getMachine(app Address) *nm.NodeMachine { | ||
| m.mutex.RLock() | ||
| defer m.mutex.Unlock() | ||
| return m.machines[app] | ||
| } | ||
|
|
||
| func closeMachines(machines map[Address]*nm.NodeMachine) (err error) { | ||
| for _, machine := range machines { | ||
| err = errors.Join(err, machine.Close()) | ||
| } | ||
| for app := range machines { | ||
| delete(machines, app) | ||
| } | ||
| return | ||
| } | ||
|
|
||
| func catchUp(ctx context.Context, | ||
| repo Repository, | ||
| app Address, | ||
| machine rm.RollupsMachine, | ||
| snapshotInputIndex *uint64, | ||
| ) (lastProcessedInputIndex *uint64, _ error) { | ||
| // A nil index indicates we should start to process from the beginning (index zero). | ||
| // A non-nil index indicates we should start to process from the next available index. | ||
| index := uint64(0) | ||
| if snapshotInputIndex != nil { | ||
| index = *snapshotInputIndex + 1 | ||
| } | ||
|
|
||
| inputs, err := repo.GetProcessedInputs(ctx, app, index) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| for _, input := range inputs { | ||
| accepted, _, _, _, err := machine.Advance(ctx, input.RawData) | ||
| _, err = nm.ToInputStatus(accepted, err) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| *lastProcessedInputIndex = input.Index | ||
| } | ||
|
|
||
| return lastProcessedInputIndex, nil | ||
| } |
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
Oops, something went wrong.