From a5bcdf14a9adca1f39bfafaa2c7ec5c95b1b6f8a Mon Sep 17 00:00:00 2001 From: danwt <30197399+danwt@users.noreply.github.com> Date: Mon, 12 Aug 2024 17:02:03 +0100 Subject: [PATCH 1/2] move ticker to submitter thread --- block/submit.go | 18 ++++++------------ 1 file changed, 6 insertions(+), 12 deletions(-) diff --git a/block/submit.go b/block/submit.go index d55423f6c..0113adee3 100644 --- a/block/submit.go +++ b/block/submit.go @@ -52,9 +52,8 @@ func SubmitLoopInner( submitter := uchannel.NewNudger() // used to avoid busy waiting (using cpu) on submitter thread eg.Go(func() error { - // 'trigger': we need one thread to continuously consume the bytes produced channel, and to monitor timer - ticker := time.NewTicker(maxBatchTime / 10) // interval does not need to match max batch time since the other thread keeps track of the actual time - defer ticker.Stop() + // 'trigger': this thread is responsible for waking up the submitter when a new block arrives, and back-pressures the block production loop + // if it gets too far ahead. for { if maxBatchSkew*maxBatchBytes < pendingBytes.Load() { // too much stuff is pending submission @@ -63,13 +62,6 @@ func SubmitLoopInner( case <-ctx.Done(): return ctx.Err() case <-trigger.C: - case <-ticker.C: - // It's theoretically possible for the thread scheduler to pause this thread after entering this if statement - // for enough time for the submitter thread to submit all the pending bytes and do the nudge, and then for the - // thread scheduler to wake up this thread after the nudge has been missed, which would be a deadlock. - // Although this is only a theoretical possibility which should never happen in practice, it may be possible, e.g. - // in adverse CPU conditions or tests using compressed timeframes. To be sound, we also nudge with the ticker, which - // has no downside. } } else { select { @@ -78,7 +70,6 @@ func SubmitLoopInner( case n := <-bytesProduced: pendingBytes.Add(uint64(n)) logger.Info("Added bytes produced to bytes pending submission counter.", "n", n) - case <-ticker.C: } } @@ -89,12 +80,14 @@ func SubmitLoopInner( }) eg.Go(func() error { - // 'submitter': this thread actually creates and submits batches + // 'submitter': this thread actually creates and submits batches, and will do it on a timer if he isn't nudged by block production timeLastSubmission := time.Now() + ticker := time.NewTicker(maxBatchTime) for { select { case <-ctx.Done(): return ctx.Err() + case <-ticker.C: case <-submitter.C: } pending := pendingBytes.Load() @@ -117,6 +110,7 @@ func SubmitLoopInner( return err } timeLastSubmission = time.Now() + ticker.Reset(maxBatchTime) pending = uatomic.Uint64Sub(&pendingBytes, nConsumed) logger.Info("Submitted a batch to both sub-layers.", "n bytes consumed from pending", nConsumed, "pending after", pending) // TODO: debug level } From 17cd13cf29b5bfd6879cc73652e656703c8f63b6 Mon Sep 17 00:00:00 2001 From: danwt <30197399+danwt@users.noreply.github.com> Date: Mon, 12 Aug 2024 17:05:30 +0100 Subject: [PATCH 2/2] small dymint refactor --- block/submit.go | 8 ++++---- block/submit_loop_test.go | 6 +++--- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/block/submit.go b/block/submit.go index 0113adee3..8c5dd5d18 100644 --- a/block/submit.go +++ b/block/submit.go @@ -72,9 +72,6 @@ func SubmitLoopInner( logger.Info("Added bytes produced to bytes pending submission counter.", "n", n) } } - - types.RollappPendingSubmissionsSkewNumBytes.Set(float64(pendingBytes.Load())) - types.RollappPendingSubmissionsSkewNumBatches.Set(float64(pendingBytes.Load() / maxBatchBytes)) submitter.Nudge() } }) @@ -82,7 +79,7 @@ func SubmitLoopInner( eg.Go(func() error { // 'submitter': this thread actually creates and submits batches, and will do it on a timer if he isn't nudged by block production timeLastSubmission := time.Now() - ticker := time.NewTicker(maxBatchTime) + ticker := time.NewTicker(maxBatchTime / 10) // interval does not need to match max batch time since we keep track anyway, it's just to wakeup for { select { case <-ctx.Done(): @@ -91,6 +88,9 @@ func SubmitLoopInner( case <-submitter.C: } pending := pendingBytes.Load() + types.RollappPendingSubmissionsSkewNumBytes.Set(float64(pendingBytes.Load())) + types.RollappPendingSubmissionsSkewNumBatches.Set(float64(pendingBytes.Load() / maxBatchBytes)) + // while there are accumulated blocks, create and submit batches!! for { done := ctx.Err() != nil diff --git a/block/submit_loop_test.go b/block/submit_loop_test.go index 8203a4536..25ee2e14f 100644 --- a/block/submit_loop_test.go +++ b/block/submit_loop_test.go @@ -99,7 +99,7 @@ func testSubmitLoopInner( nProducedBytes.Add(^uint64(consumed - 1)) // subtract timeLastProgressT := time.Unix(timeLastProgress.Load(), 0) - absoluteMax := int64(1.5 * float64(args.maxTime)) // allow some leeway for code execution + absoluteMax := int64(2 * float64(args.maxTime)) // allow some leeway for code execution. Tests may run on small boxes (GH actions) timeSinceLast := time.Since(timeLastProgressT).Milliseconds() require.True(t, timeSinceLast < absoluteMax, "too long since last update", "timeSinceLast", timeSinceLast, "max", absoluteMax) @@ -115,7 +115,7 @@ func TestSubmitLoopFastProducerHaltingSubmitter(t *testing.T) { testSubmitLoop( t, testArgs{ - nParallel: 100, + nParallel: 50, testDuration: 2 * time.Second, batchSkew: 10, batchBytes: 100, @@ -136,7 +136,7 @@ func TestSubmitLoopTimer(t *testing.T) { testSubmitLoop( t, testArgs{ - nParallel: 100, + nParallel: 50, testDuration: 2 * time.Second, batchSkew: 10, batchBytes: 100,