Skip to content

Commit 1216dc1

Browse files
rodrodrosrodrodros
authored
feat: add throtled compiler to negate resource starvation (#3744)
* feat: add throtled compiler to negate resource starvation * test(node): throttled compiler * test(node): throttled test with synctest
1 parent d7a0bb4 commit 1216dc1

16 files changed

Lines changed: 192 additions & 38 deletions

‎cmd/juno/juno.go‎

Lines changed: 15 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -106,6 +106,7 @@ const (
106106
dbCompressionF = "db-compression"
107107
rpcRequestTimeoutF = "rpc-request-timeout"
108108
maxConcurrentCompilationsF = "max-concurrent-compilations"
109+
maxCompilationQueueF = "max-compilation-queue"
109110
maxCompilationMemoryF = "max-compilation-memory"
110111
maxCompilationCPUTimeF = "max-compilation-cpu-time"
111112
disableReceivedTxnStreamF = "disable-received-txn-stream"
@@ -255,7 +256,9 @@ const (
255256
"Use zstd for low storage."
256257
rpcRequestTimeoutUsage = "Maximum time for an RPC request to complete."
257258
maxConcurrentCompilationsUsage = "Maximum concurrent Sierra compilations."
258-
maxCompilationMemoryUsage = "Maximum memory (in MB) each Sierra compilation process may " +
259+
maxCompilationQueueUsage = "Maximum number of compilation requests to queue after " +
260+
"reaching max-concurrent-compilations before starting to reject incoming requests."
261+
maxCompilationMemoryUsage = "Maximum memory (in MB) each Sierra compilation process may " +
259262
"use; a compilation exceeding it is aborted. Enforced on Linux only. 0 disables the limit."
260263
maxCompilationCPUTimeUsage = "Maximum CPU time (in seconds) each Sierra compilation process " +
261264
"may consume; a compilation exceeding it is aborted. Enforced on Linux only. " +
@@ -575,6 +578,11 @@ func NewCmd(config *node.Config, run func(*cobra.Command, []string) error) *cobr
575578
uint(defaultMaxConcurrentCompilations),
576579
maxConcurrentCompilationsUsage,
577580
)
581+
junoCmd.Flags().Uint(
582+
maxCompilationQueueF,
583+
2*uint(defaultMaxConcurrentCompilations),
584+
maxCompilationQueueUsage,
585+
)
578586
junoCmd.Flags().Uint(maxCompilationMemoryF, defaultMaxCompilationMemory, maxCompilationMemoryUsage)
579587
junoCmd.Flags().Uint(
580588
maxCompilationCPUTimeF, defaultMaxCompilationCPUTime, maxCompilationCPUTimeUsage,
@@ -583,8 +591,12 @@ func NewCmd(config *node.Config, run func(*cobra.Command, []string) error) *cobr
583591
versionedConstantsFileF, defaultVersionedConstantsFile, versionedConstantsFileUsage,
584592
)
585593
setCategory(junoCmd, catVMCompile,
586-
maxVMsF, maxVMQueueF, maxConcurrentCompilationsF,
587-
maxCompilationMemoryF, maxCompilationCPUTimeF, versionedConstantsFileF,
594+
maxVMsF, maxVMQueueF,
595+
maxConcurrentCompilationsF,
596+
maxCompilationQueueF,
597+
maxCompilationMemoryF,
598+
maxCompilationCPUTimeF,
599+
versionedConstantsFileF,
588600
)
589601

590602
// --- Custom Network ---

‎cmd/juno/juno_test.go‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,7 @@ func TestConfigPrecedence(t *testing.T) {
7373
defaultSubmittedTransactionsCacheEntryTTL := 5 * time.Minute
7474
defaultRPCRequestTimeout := 1 * time.Minute
7575
defaultMaxConcurrentCompilations := uint(runtime.GOMAXPROCS(0))
76+
defaultMaxCompilationQueue := 2 * defaultMaxConcurrentCompilations
7677
defaultMaxCompilationMemory := uint(4 * 1024)
7778
defaultMaxCompilationCPUTime := uint(10)
7879
if runtime.GOOS != "linux" {
@@ -125,6 +126,7 @@ func TestConfigPrecedence(t *testing.T) {
125126
ReadinessBlockTolerance: 6,
126127
RPCRequestTimeout: defaultRPCRequestTimeout,
127128
MaxConcurrentCompilations: defaultMaxConcurrentCompilations,
129+
MaxCompilationQueue: defaultMaxCompilationQueue,
128130
MaxCompilationMemory: defaultMaxCompilationMemory,
129131
MaxCompilationCPUTime: defaultMaxCompilationCPUTime,
130132
PruneMinAge: defaultPruneMinAge,
@@ -172,6 +174,7 @@ func TestConfigPrecedence(t *testing.T) {
172174
ReadinessBlockTolerance: 6,
173175
RPCRequestTimeout: defaultRPCRequestTimeout,
174176
MaxConcurrentCompilations: defaultMaxConcurrentCompilations,
177+
MaxCompilationQueue: defaultMaxCompilationQueue,
175178
MaxCompilationMemory: defaultMaxCompilationMemory,
176179
MaxCompilationCPUTime: defaultMaxCompilationCPUTime,
177180
PruneMinAge: defaultPruneMinAge,
@@ -293,6 +296,7 @@ pprof: true
293296
ReadinessBlockTolerance: 6,
294297
RPCRequestTimeout: defaultRPCRequestTimeout,
295298
MaxConcurrentCompilations: defaultMaxConcurrentCompilations,
299+
MaxCompilationQueue: defaultMaxCompilationQueue,
296300
MaxCompilationMemory: defaultMaxCompilationMemory,
297301
MaxCompilationCPUTime: defaultMaxCompilationCPUTime,
298302
PruneMinAge: defaultPruneMinAge,
@@ -346,6 +350,7 @@ http-port: 4576
346350
ReadinessBlockTolerance: 6,
347351
RPCRequestTimeout: defaultRPCRequestTimeout,
348352
MaxConcurrentCompilations: defaultMaxConcurrentCompilations,
353+
MaxCompilationQueue: defaultMaxCompilationQueue,
349354
MaxCompilationMemory: defaultMaxCompilationMemory,
350355
MaxCompilationCPUTime: defaultMaxCompilationCPUTime,
351356
PruneMinAge: defaultPruneMinAge,
@@ -398,6 +403,7 @@ http-port: 4576
398403
ReadinessBlockTolerance: 6,
399404
RPCRequestTimeout: defaultRPCRequestTimeout,
400405
MaxConcurrentCompilations: defaultMaxConcurrentCompilations,
406+
MaxCompilationQueue: defaultMaxCompilationQueue,
401407
MaxCompilationMemory: defaultMaxCompilationMemory,
402408
MaxCompilationCPUTime: defaultMaxCompilationCPUTime,
403409
PruneMinAge: defaultPruneMinAge,
@@ -450,6 +456,7 @@ http-port: 4576
450456
ReadinessBlockTolerance: 6,
451457
RPCRequestTimeout: defaultRPCRequestTimeout,
452458
MaxConcurrentCompilations: defaultMaxConcurrentCompilations,
459+
MaxCompilationQueue: defaultMaxCompilationQueue,
453460
MaxCompilationMemory: defaultMaxCompilationMemory,
454461
MaxCompilationCPUTime: defaultMaxCompilationCPUTime,
455462
PruneMinAge: defaultPruneMinAge,
@@ -528,6 +535,7 @@ db-cache-size: 1024
528535
ReadinessBlockTolerance: 6,
529536
RPCRequestTimeout: defaultRPCRequestTimeout,
530537
MaxConcurrentCompilations: defaultMaxConcurrentCompilations,
538+
MaxCompilationQueue: defaultMaxCompilationQueue,
531539
MaxCompilationMemory: defaultMaxCompilationMemory,
532540
MaxCompilationCPUTime: defaultMaxCompilationCPUTime,
533541
PruneMinAge: defaultPruneMinAge,
@@ -583,6 +591,7 @@ network: sepolia
583591
ReadinessBlockTolerance: 6,
584592
RPCRequestTimeout: defaultRPCRequestTimeout,
585593
MaxConcurrentCompilations: defaultMaxConcurrentCompilations,
594+
MaxCompilationQueue: defaultMaxCompilationQueue,
586595
MaxCompilationMemory: defaultMaxCompilationMemory,
587596
MaxCompilationCPUTime: defaultMaxCompilationCPUTime,
588597
PruneMinAge: defaultPruneMinAge,
@@ -634,6 +643,7 @@ network: sepolia
634643
ReadinessBlockTolerance: 6,
635644
RPCRequestTimeout: defaultRPCRequestTimeout,
636645
MaxConcurrentCompilations: defaultMaxConcurrentCompilations,
646+
MaxCompilationQueue: defaultMaxCompilationQueue,
637647
MaxCompilationMemory: defaultMaxCompilationMemory,
638648
MaxCompilationCPUTime: defaultMaxCompilationCPUTime,
639649
PruneMinAge: defaultPruneMinAge,
@@ -683,6 +693,7 @@ network: sepolia
683693
ReadinessBlockTolerance: 6,
684694
RPCRequestTimeout: defaultRPCRequestTimeout,
685695
MaxConcurrentCompilations: defaultMaxConcurrentCompilations,
696+
MaxCompilationQueue: defaultMaxCompilationQueue,
686697
MaxCompilationMemory: defaultMaxCompilationMemory,
687698
MaxCompilationCPUTime: defaultMaxCompilationCPUTime,
688699
PruneMinAge: defaultPruneMinAge,
@@ -733,6 +744,7 @@ network: sepolia
733744
ReadinessBlockTolerance: 6,
734745
RPCRequestTimeout: defaultRPCRequestTimeout,
735746
MaxConcurrentCompilations: defaultMaxConcurrentCompilations,
747+
MaxCompilationQueue: defaultMaxCompilationQueue,
736748
MaxCompilationMemory: defaultMaxCompilationMemory,
737749
MaxCompilationCPUTime: defaultMaxCompilationCPUTime,
738750
PruneMinAge: defaultPruneMinAge,
@@ -782,6 +794,7 @@ network: sepolia
782794
ReadinessBlockTolerance: 6,
783795
RPCRequestTimeout: defaultRPCRequestTimeout,
784796
MaxConcurrentCompilations: defaultMaxConcurrentCompilations,
797+
MaxCompilationQueue: defaultMaxCompilationQueue,
785798
MaxCompilationMemory: defaultMaxCompilationMemory,
786799
MaxCompilationCPUTime: defaultMaxCompilationCPUTime,
787800
PruneMinAge: defaultPruneMinAge,
@@ -833,6 +846,7 @@ network: sepolia
833846
ReadinessBlockTolerance: 6,
834847
RPCRequestTimeout: defaultRPCRequestTimeout,
835848
MaxConcurrentCompilations: defaultMaxConcurrentCompilations,
849+
MaxCompilationQueue: defaultMaxCompilationQueue,
836850
MaxCompilationMemory: defaultMaxCompilationMemory,
837851
MaxCompilationCPUTime: defaultMaxCompilationCPUTime,
838852
PruneMinAge: defaultPruneMinAge,

‎docs/generate-config.js‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -109,6 +109,9 @@ const extractConfigs = (codebase) => {
109109
if (configName === "max-concurrent-compilations") {
110110
defaultValue = "CPU Cores";
111111
}
112+
if (configName === "max-compilation-queue") {
113+
defaultValue = "2 * max-concurrent-compilations";
114+
}
112115
if (configName === "gw-timeouts") {
113116
defaultValue = "5s";
114117
}

‎node/metrics.go‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -363,6 +363,24 @@ func makeVMThrottlerMetrics(throttledVM *ThrottledVM) {
363363
prometheus.MustRegister(vmJobs, vmQueue)
364364
}
365365

366+
func makeCompilerThrottlerMetrics(throttledCompiler *ThrottledCompiler) {
367+
compilerJobs := prometheus.NewGaugeFunc(prometheus.GaugeOpts{
368+
Namespace: "compiler",
369+
Name: "jobs",
370+
Help: "Number of currently running compiler jobs",
371+
}, func() float64 {
372+
return float64(throttledCompiler.JobsRunning())
373+
})
374+
compilerQueue := prometheus.NewGaugeFunc(prometheus.GaugeOpts{
375+
Namespace: "compiler",
376+
Name: "queue",
377+
Help: "Number of compiler jobs waiting in the queue",
378+
}, func() float64 {
379+
return float64(throttledCompiler.QueueLen())
380+
})
381+
prometheus.MustRegister(compilerJobs, compilerQueue)
382+
}
383+
366384
func makePrunerMetrics() pruner.EventListener {
367385
oldestBlockKept := prometheus.NewGauge(prometheus.GaugeOpts{
368386
Namespace: namespacePruner,

‎node/node.go‎

Lines changed: 17 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -135,6 +135,7 @@ type Config struct {
135135

136136
RPCRequestTimeout time.Duration `mapstructure:"rpc-request-timeout"`
137137
MaxConcurrentCompilations uint `mapstructure:"max-concurrent-compilations"`
138+
MaxCompilationQueue uint `mapstructure:"max-compilation-queue"`
138139
MaxCompilationMemory uint `mapstructure:"max-compilation-memory"` // megabytes
139140
MaxCompilationCPUTime uint `mapstructure:"max-compilation-cpu-time"` // CPU seconds
140141
NewState bool `mapstructure:"new-state"`
@@ -290,14 +291,17 @@ func New(cfg *Config, version string, logLevel *log.Level) (*Node, error) {
290291
var nodeVM vm.VM
291292
var throttledVM *ThrottledVM
292293

293-
compiler := compiler.New(
294-
&compiler.Config{
295-
MaxConcurrent: cfg.MaxConcurrentCompilations,
296-
MaxMemory: uint64(cfg.MaxCompilationMemory) * 1024 * 1024,
297-
MaxCPUTime: uint64(cfg.MaxCompilationCPUTime),
298-
},
299-
"",
300-
logger,
294+
throttledCompiler := NewThrottledCompiler(
295+
compiler.New(
296+
&compiler.Config{
297+
MaxMemory: uint64(cfg.MaxCompilationMemory) * 1024 * 1024,
298+
MaxCPUTime: uint64(cfg.MaxCompilationCPUTime),
299+
},
300+
"",
301+
logger,
302+
),
303+
cfg.MaxConcurrentCompilations,
304+
int32(cfg.MaxCompilationQueue),
301305
)
302306

303307
if cfg.Sequencer {
@@ -324,7 +328,7 @@ func New(cfg *Config, version string, logLevel *log.Level) (*Node, error) {
324328
pKey, time.Second*time.Duration(cfg.SeqBlockTime), logger)
325329
seq.WithPlugin(junoPlugin)
326330
rpcHandler = rpc.New(chain, &seq, throttledVM, version, logger, &cfg.Network).
327-
WithCompiler(compiler).
331+
WithCompiler(throttledCompiler).
328332
WithMempool(mempool).
329333
WithCallMaxSteps(cfg.RPCCallMaxSteps).
330334
WithCallMaxGas(cfg.RPCCallMaxGas)
@@ -415,7 +419,7 @@ func New(cfg *Config, version string, logLevel *log.Level) (*Node, error) {
415419
&cfg.Network,
416420
logger,
417421
database,
418-
compiler,
422+
throttledCompiler,
419423
)
420424
if err != nil {
421425
return nil, fmt.Errorf("set up p2p service: %w", err)
@@ -435,7 +439,7 @@ func New(cfg *Config, version string, logLevel *log.Level) (*Node, error) {
435439
)
436440
services = append(services, submittedTransactionsCache)
437441
rpcHandler = rpc.New(chain, syncReader, throttledVM, version, logger, &cfg.Network).
438-
WithCompiler(compiler).
442+
WithCompiler(throttledCompiler).
439443
WithGateway(gatewayClient).
440444
WithFeeder(client).
441445
WithSubmittedTransactionsCache(submittedTransactionsCache).
@@ -559,6 +563,7 @@ func New(cfg *Config, version string, logLevel *log.Level) (*Node, error) {
559563
if cfg.Metrics {
560564
makeJeMallocMetrics()
561565
makeVMThrottlerMetrics(throttledVM)
566+
makeCompilerThrottlerMetrics(throttledCompiler)
562567
makePebbleMetrics(database)
563568
makeJunoMetrics(version)
564569
database.WithListener(makeDBMetrics())
@@ -591,7 +596,7 @@ func New(cfg *Config, version string, logLevel *log.Level) (*Node, error) {
591596
version: version,
592597
db: database,
593598
blockchain: chain,
594-
compiler: compiler,
599+
compiler: throttledCompiler,
595600
services: services,
596601
earlyServices: earlyServices,
597602
}

‎node/throttled_compiler.go‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,35 @@
1+
package node
2+
3+
import (
4+
"context"
5+
6+
"github.com/NethermindEth/juno/starknet"
7+
"github.com/NethermindEth/juno/starknet/compiler"
8+
"github.com/NethermindEth/juno/utils"
9+
)
10+
11+
var _ compiler.Compiler = (*ThrottledCompiler)(nil)
12+
13+
type ThrottledCompiler struct {
14+
*utils.Throttler[compiler.Compiler]
15+
}
16+
17+
func NewThrottledCompiler(
18+
res compiler.Compiler, concurrencyBudget uint, maxQueueLen int32,
19+
) *ThrottledCompiler {
20+
return &ThrottledCompiler{
21+
Throttler: utils.NewThrottler(concurrencyBudget, &res).WithMaxQueueLen(maxQueueLen),
22+
}
23+
}
24+
25+
func (tc *ThrottledCompiler) Compile(
26+
ctx context.Context, sierra *starknet.SierraClass,
27+
) (*starknet.CasmClass, error) {
28+
var result *starknet.CasmClass
29+
err := tc.Do(func(c *compiler.Compiler) error {
30+
var cErr error
31+
result, cErr = (*c).Compile(ctx, sierra)
32+
return cErr
33+
})
34+
return result, err
35+
}

‎node/throttled_compiler_test.go‎

Lines changed: 63 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,63 @@
1+
package node_test
2+
3+
import (
4+
"context"
5+
"sync"
6+
"testing"
7+
"testing/synctest"
8+
9+
"github.com/NethermindEth/juno/node"
10+
"github.com/NethermindEth/juno/starknet"
11+
"github.com/NethermindEth/juno/utils"
12+
"github.com/stretchr/testify/assert"
13+
"github.com/stretchr/testify/require"
14+
)
15+
16+
// blockingCompiler is a stub compiler.Compiler whose Compile blocks until
17+
// waitOn receives, letting the test hold compilation slots open.
18+
type blockingCompiler struct {
19+
waitOn chan struct{}
20+
}
21+
22+
func (b *blockingCompiler) Compile(
23+
_ context.Context, _ *starknet.SierraClass,
24+
) (*starknet.CasmClass, error) {
25+
<-b.waitOn
26+
return &starknet.CasmClass{}, nil
27+
}
28+
29+
func TestThrottledCompiler(t *testing.T) {
30+
synctest.Test(t, func(t *testing.T) {
31+
waitOn := make(chan struct{})
32+
// Concurrency budget of 2 and a max queue of 2.
33+
throttled := node.NewThrottledCompiler(&blockingCompiler{waitOn: waitOn}, 2, 2)
34+
35+
var wg sync.WaitGroup
36+
// compile spawns a Compile call, then waits for the bubble to settle so
37+
// the throttler's running/queued counts are observed deterministically.
38+
compile := func(wantRunning, wantQueued int) {
39+
wg.Go(func() {
40+
_, err := throttled.Compile(t.Context(), &starknet.SierraClass{})
41+
assert.NoError(t, err) // assert, not require: runs off the test goroutine
42+
})
43+
synctest.Wait() // block until the spawned goroutine is durably blocked
44+
assert.Equal(t, wantRunning, throttled.JobsRunning())
45+
assert.Equal(t, wantQueued, throttled.QueueLen())
46+
}
47+
48+
compile(1, 0) // occupies a slot
49+
compile(2, 0) // occupies the other slot
50+
compile(2, 1) // queued
51+
compile(2, 2) // queued
52+
53+
// The queue is full, so the next request is rejected.
54+
_, err := throttled.Compile(t.Context(), &starknet.SierraClass{})
55+
require.ErrorIs(t, err, utils.ErrResourceBusy)
56+
57+
// Release the four running/queued jobs and let them finish.
58+
for range 4 {
59+
waitOn <- struct{}{}
60+
}
61+
wg.Wait()
62+
})
63+
}

‎rpc/rpccore/rpccore.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@ const (
1515
MaxEventFilterKeys = 1024
1616
TraceCacheSize = 128
1717
ThrottledVMErr = "VM throughput limit reached"
18+
ThrottledCompilerErr = "Compiler throughput limit reached"
1819
MaxBlocksBack = 1024
1920
EntrypointNotFoundFelt string = "0x454e545259504f494e545f4e4f545f464f554e44"
2021
ErrEPSNotFound = "Entry point EntryPointSelector(%s) not found in contract."

‎rpc/v10/simulation.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -319,6 +319,9 @@ func (h *Handler) prepareTransactions(
319319
network,
320320
)
321321
if aErr != nil {
322+
if errors.Is(aErr, utils.ErrResourceBusy) {
323+
return nil, nil, rpccore.ErrInternal.CloneWithData(rpccore.ThrottledCompilerErr)
324+
}
322325
return nil, nil, jsonrpc.Err(jsonrpc.InvalidParams, aErr.Error())
323326
}
324327

‎rpc/v10/transaction.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -200,6 +200,9 @@ func (h *Handler) addToMempool(
200200
ctx, h.compiler, tx, h.bcReader.Network(),
201201
)
202202
if err != nil {
203+
if errors.Is(err, utils.ErrResourceBusy) {
204+
return AddTxResponse{}, rpccore.ErrInternal.CloneWithData(rpccore.ThrottledCompilerErr)
205+
}
203206
return AddTxResponse{}, rpccore.ErrInternal.CloneWithData(err.Error())
204207
}
205208
if err = h.memPool.Push(ctx, &mempool.BroadcastedTransaction{

0 commit comments

Comments
 (0)