Skip to content

Commit

Permalink
feat: add the machines package
Browse files Browse the repository at this point in the history
  • Loading branch information
renan061 committed Aug 30, 2024
1 parent 37328f2 commit 08d1993
Show file tree
Hide file tree
Showing 8 changed files with 397 additions and 7 deletions.
51 changes: 50 additions & 1 deletion docs/config.md
Original file line number Diff line number Diff line change
Expand Up @@ -223,6 +223,13 @@ When enabled, will connect to postgres database via SSL.
* **Type:** `bool`
* **Default:** `"true"`

## `CARTESI_ADVANCER_POLLING_INTERVAL`

How many seconds the node will wait before querying the database for new inputs.

* **Type:** `Duration`
* **Default:** `"30"`

## `CARTESI_EPOCH_LENGTH`

Length of a rollups epoch in blocks.
Expand All @@ -234,7 +241,7 @@ At the end of each epoch, the node will send claims to the blockchain.

## `CARTESI_EVM_READER_RETRY_POLICY_MAX_DELAY`

How seconds the retry policy will wait between retries.
How many seconds the retry policy will wait between retries.

* **Type:** `Duration`
* **Default:** `"3"`
Expand All @@ -246,6 +253,13 @@ How many times some functions should be retried after an error.
* **Type:** `uint64`
* **Default:** `"3"`

## `CARTESI_MAX_CONCURRENT_INSPECTS`

Maximum number of inspect-state requests that can be concurrently active.

* **Type:** `uint8`
* **Default:** `"10"`

## `CARTESI_VALIDATOR_POLLING_INTERVAL`

How many seconds the node will wait before trying to finish epochs for all applications.
Expand All @@ -258,3 +272,38 @@ How many seconds the node will wait before trying to finish epochs for all appli
Path to the directory with the cartesi-machine snapshot that will be loaded by the node.

* **Type:** `string`

## `CARTESI_MACHINE_ADVANCE_TIMEOUT`

TODO.

* **Type:** `Duration`
* **Default:** `"60"`

## `CARTESI_MACHINE_INC_CYCLES`

TODO.

* **Type:** `uint64`
* **Default:** `"50000000"`

## `CARTESI_MACHINE_INSPECT_TIMEOUT`

TODO.

* **Type:** `Duration`
* **Default:** `"10"`

## `CARTESI_MACHINE_MAX_CYCLES`

TODO.

* **Type:** `uint64`
* **Default:** `"5000000000"`

## `CARTESI_MACHINE_SERVER_VERBOSITY`

TODO.

* **Type:** `string`
* **Default:** `"info"`
167 changes: 167 additions & 0 deletions internal/node/advancer/machines/machines.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,167 @@
// (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"

"github.com/cartesi/rollups-node/internal/node/config"
"github.com/cartesi/rollups-node/internal/node/model"
"github.com/cartesi/rollups-node/internal/nodemachine"
"github.com/cartesi/rollups-node/internal/repository"
"github.com/cartesi/rollups-node/pkg/emulator"
"github.com/cartesi/rollups-node/pkg/rollupsmachine"
"github.com/cartesi/rollups-node/pkg/rollupsmachine/cartesimachine"
)

type Repository interface {
GetAppData(context.Context) ([]*repository.AppData, error)

Check failure on line 24 in internal/node/advancer/machines/machines.go

View workflow job for this annotation

GitHub Actions / test-go

undefined: repository.AppData (typecheck)
}

type AdvanceMachine interface {
Advance(_ context.Context, input []byte, index uint64) (*nodemachine.AdvanceResult, error)
}

type InspectMachine interface {
Inspect(_ context.Context, query []byte) (*nodemachine.InspectResult, error)
}

type Machines struct {
mutex sync.RWMutex
machines map[model.Address]*nodemachine.NodeMachine
}

func Load(ctx context.Context, config config.NodeConfig, repo Repository) (*Machines, error) {
appData, err := repo.GetAppData(ctx)
if err != nil {
return nil, err
}

machines := map[model.Address]*nodemachine.NodeMachine{}

maxConcurrentInspects := config.MaxConcurrentInspects

serverVerbosity := config.MachineServerVerbosity
machineInc := config.MachineIncCycles
machineMax := config.MachineMaxCycles
machineAdvanceTimeout := config.MachineAdvanceTimeout
machineInspectTimeout := config.MachineInspectTimeout

for _, appData := range appData {
appAddress := appData.AppAddress
snapshotPath := appData.SnapshotPath
snapshotInputIndex := appData.InputIndex

address, err := cartesimachine.StartServer(serverVerbosity, 0, os.Stdout, os.Stderr)
if err != nil {
return nil, closeMachines(machines)
}

config := &emulator.MachineRuntimeConfig{}
cartesiMachine, err := cartesimachine.Load(ctx, snapshotPath, address, config)
if err != nil {
err = errors.Join(err, cartesimachine.StopServer(address), closeMachines(machines))
return nil, err
}

rollupsMachine, err := rollupsmachine.New(ctx, cartesiMachine, machineInc, machineMax)
if err != nil {
err = errors.Join(err, cartesiMachine.Close(ctx), closeMachines(machines))
return nil, err
}

nodeMachine, err := nodemachine.New(rollupsMachine,
snapshotInputIndex,
machineAdvanceTimeout,
machineInspectTimeout,
maxConcurrentInspects)
if err != nil {
err = errors.Join(err, rollupsMachine.Close(ctx), closeMachines(machines))
return nil, err
}

machines[appAddress] = nodeMachine
}

return &Machines{machines: machines}, nil
}

func (m *Machines) GetAdvanceMachine(app model.Address) AdvanceMachine {
return m.get(app)
}

func (m *Machines) GetInspectMachine(app model.Address) InspectMachine {
return m.get(app)
}

func (m *Machines) Set(app model.Address, machine *nodemachine.NodeMachine) bool {
m.mutex.Lock()
defer m.mutex.Unlock()

if _, ok := m.machines[app]; ok {
return false
} else {
m.machines[app] = machine
return true
}
}

func (m *Machines) Remove(app model.Address) *nodemachine.NodeMachine {
m.mutex.Lock()
defer m.mutex.Unlock()

if machine, ok := m.machines[app]; ok {
return nil
} else {
delete(m.machines, app)
return machine
}
}

func (m *Machines) Keys() []model.Address {
m.mutex.RLock()
defer m.mutex.Unlock()

keys := make([]model.Address, len(m.machines))
i := 0
for k := range m.machines {
keys[i] = k
i++
}
return keys
}

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) get(app model.Address) *nodemachine.NodeMachine {
m.mutex.RLock()
defer m.mutex.Unlock()
return m.machines[app]
}

func closeMachines(machines map[model.Address]*nodemachine.NodeMachine) (err error) {
for _, machine := range machines {
err = errors.Join(err, machine.Close())
}
for app := range machines {
delete(machines, app)
}
return
}
18 changes: 18 additions & 0 deletions internal/node/config/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,8 @@ package config
import (
"fmt"
"os"

"github.com/cartesi/rollups-node/pkg/rollupsmachine/cartesimachine"
)

// NodeConfig contains all the Node variables.
Expand Down Expand Up @@ -39,7 +41,15 @@ type NodeConfig struct {
ExperimentalSunodoValidatorEnabled bool
ExperimentalSunodoValidatorRedisEndpoint string
Auth Auth
MaxConcurrentInspects uint8
AdvancerPollingInterval Duration
ValidatorPollingInterval Duration
// Temporary
MachineServerVerbosity cartesimachine.ServerVerbosity
MachineIncCycles uint64
MachineMaxCycles uint64
MachineAdvanceTimeout Duration
MachineInspectTimeout Duration
}

// Auth is used to sign transactions.
Expand Down Expand Up @@ -106,7 +116,15 @@ func FromEnv() NodeConfig {
if getFeatureClaimerEnabled() && !getExperimentalSunodoValidatorEnabled() {
config.Auth = authFromEnv()
}
config.MaxConcurrentInspects = getMaxConcurrentInspects()
config.AdvancerPollingInterval = getAdvancerPollingInterval()
config.ValidatorPollingInterval = getValidatorPollingInterval()
// Temporary.
config.MachineServerVerbosity = cartesimachine.ServerVerbosity(getMachineServerVerbosity())
config.MachineIncCycles = getMachineIncCycles()
config.MachineMaxCycles = getMachineMaxCycles()
config.MachineAdvanceTimeout = getMachineAdvanceTimeout()
config.MachineInspectTimeout = getMachineInspectTimeout()
return config
}

Expand Down
48 changes: 47 additions & 1 deletion internal/node/config/generate/Config.toml
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,19 @@ How many times some functions should be retried after an error."""
default = "3"
go-type = "Duration"
description = """
How seconds the retry policy will wait between retries."""
How many seconds the retry policy will wait between retries."""

[rollups.CARTESI_MAX_CONCURRENT_INSPECTS]
default = "10"
go-type = "uint8"
description = """
Maximum number of inspect-state requests that can be concurrently active."""

[rollups.CARTESI_ADVANCER_POLLING_INTERVAL]
default = "30"
go-type = "Duration"
description = """
How many seconds the node will wait before querying the database for new inputs."""

[rollups.CARTESI_VALIDATOR_POLLING_INTERVAL]
default = "30"
Expand Down Expand Up @@ -250,3 +262,37 @@ go-type = "bool"
description = """
When enabled, prints server-manager output to stdout and stderr directly.
All other log configurations are ignored."""

#
# Temporary
#

[temp.CARTESI_MACHINE_SERVER_VERBOSITY]
default = "info"
go-type = "string"
description = """
TODO."""

[temp.CARTESI_MACHINE_INC_CYCLES]
default = "50000000"
go-type = "uint64"
description = """
TODO."""

[temp.CARTESI_MACHINE_MAX_CYCLES]
default = "5000000000"
go-type = "uint64"
description = """
TODO."""

[temp.CARTESI_MACHINE_ADVANCE_TIMEOUT]
default = "60"
go-type = "Duration"
description = """
TODO."""

[temp.CARTESI_MACHINE_INSPECT_TIMEOUT]
default = "10"
go-type = "Duration"
description = """
TODO."""
12 changes: 12 additions & 0 deletions internal/node/config/generate/code.go
Original file line number Diff line number Diff line change
Expand Up @@ -100,6 +100,16 @@ func ToInt64FromString(s string) (int64, error) {
return strconv.ParseInt(s, 10, 64)
}
func ToUint8FromString(s string) (uint8, error) {
value, err := strconv.ParseUint(s, 10, 8)
return uint8(value), err
}
func ToUint32FromString(s string) (uint32, error) {
value, err := strconv.ParseUint(s, 10, 32)
return uint32(value), err
}
func ToUint64FromString(s string) (uint64, error) {
value, err := strconv.ParseUint(s, 10, 64)
return value, err
Expand Down Expand Up @@ -164,6 +174,8 @@ var (
toBool = strconv.ParseBool
toInt = strconv.Atoi
toInt64 = ToInt64FromString
toUint8 = ToUint8FromString
toUint32 = ToUint32FromString
toUint64 = ToUint64FromString
toString = ToStringFromString
toDuration = ToDurationFromSeconds
Expand Down
Loading

0 comments on commit 08d1993

Please sign in to comment.