-
Notifications
You must be signed in to change notification settings - Fork 175
feat(flow): implement adaptive throttler #196
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Merged
Merged
Changes from 13 commits
Commits
Show all changes
68 commits
Select commit
Hold shift + click to select a range
b373d2a
feat(flow): implement adaptive throttler with resource monitoring and…
kxrxh bf9fa92
docs(README): add AdaptiveThrottler description to documentation
kxrxh 55c5fe1
fix(flow): emit partial sliding window on early close (#192)
reugn 2bc6a86
fix: resolve all golangci-lint issues
kxrxh fbc0df7
refactor(adaptive_throttler): streamline main function by removing co…
kxrxh 0d97643
refactor: rename CPUUsageModeReal to CPUUsageModeMeasured
kxrxh 837e3d6
feat(adaptive_throttler): enhance resource monitoring for containeriz…
kxrxh edfb728
Merge pull request #1 from reugn/master
kxrxh fa6e81f
feat(adaptive_throttler): add demo for adaptive throttling pipeline
kxrxh d1d210c
refactor(adaptive_throttler): update config validation message
kxrxh 238ca8b
docs(README): fix punctuation in AdaptiveThrottler description
kxrxh 5c7bc28
refactor(adaptive_throttler): rename resourceMonitorInterface to reso…
kxrxh 754d39e
refactor(resource_monitor): move stats collection to initSampler
kxrxh 135139f
refactor(adaptive_throttler): extract system monitoring into internal…
kxrxh a224484
refactor(sysmonitor): improve CPU time retrieval for current process
kxrxh 04586f7
feat(adaptive_throttler): enhance demo with CPU workload simulation
kxrxh 6ec1760
docs(adaptive_throttler): add clarification comment for constrained v…
kxrxh 0633447
fix(sysmonitor): add safety checks for CPU count and handle memory st…
kxrxh fe90168
refactor(sysmonitor): unify memory reading functions for cgroup v1 an…
kxrxh 5ab4ea1
refactor(sysmonitor): streamline memory reading logic and remove unus…
kxrxh a92689d
refactor(sysmonitor): additional context for errors was added
kxrxh c460ee3
refactor(sysmonitor): remove outdated validation comment in getProces…
kxrxh 4eb9507
fix(sysmonitor): add overflow check for memory value conversion
kxrxh 2e8e86a
fix(sysmonitor): enhance error handling in memory status retrieval
kxrxh 5df55d2
docs(adaptive_throttler): update demo comments to clarify adaptive th…
kxrxh aaf9377
refactor(resource_monitor_test): remove unnecessary helper call in te…
kxrxh 9b7cba6
Merge branch 'feature/adaptive-throttler' of https://github.com/kxrxh…
kxrxh a437518
refactor(resource_monitor): simplify comment for division by zero check
kxrxh 08f6012
chore(dependencies): downgrade Go version to 1.23.0 and update indire…
kxrxh e7a1c81
feat(adaptive_throttler): enforce minimum sample interval to reduce C…
kxrxh 226098e
fix(adaptive_throttler): prevent negative target rate in rate adaptation
kxrxh 0707bb0
docs(adaptive_throttler): clarify comment on MemoryReader function re…
kxrxh df7560a
docs(adaptive_throttler): format comment for CPU usage sampling modes
kxrxh 2964497
feat(resource_monitor): add comment to indicate start of periodic res…
kxrxh a5fb0d4
docs(adaptive_throttler): update MemoryReader comment for clarity on …
kxrxh 95ce780
chore(dependencies): remove go.sum and update go.mod
kxrxh 8fa0e0f
refactor(resource_monitor): move clampPercent and validatePercent fun…
kxrxh 77b55f2
refactor(flow): rename util.go to operators.go for better clarity
kxrxh 9b5f438
refactor(adaptive_throttler): remove unnecessary pointer dereference …
kxrxh 55ca0c5
refactor(resource_monitor): change stats from atomic.Value to atomic.…
kxrxh d1c811e
refactor(adaptive_throttler): remove unnecessary mutex for rate adapt…
kxrxh 5a46a4f
refactor(adaptive_throttler): reorder functions and improved adaptRat…
kxrxh e097d6b
feat(adaptive_throttler): add MaxBufferSize configuration and validat…
kxrxh cd8ad61
refactor(sysmonitor): simplify memory initialization by removing unne…
kxrxh 809f604
fix(sysmonitor): ensure available memory does not exceed total memory
kxrxh 904d9e9
refactor(adaptive_throttler): fix golangci-lint errors
kxrxh 60b20a4
refactor(adaptive_throttler): change validateConfig function to a met…
kxrxh ab90a2c
refactor(adaptive_throttler, resource_monitor): enhance configuration…
kxrxh 7d4abd4
test: add comprehensive test coverage for adaptive throttler and syst…
kxrxh 604b5b5
test(resource_monitor): add tests for clamping and validation of perc…
kxrxh 7dfff93
test(sysmonitor): refactor memory tests to use a helper function to f…
kxrxh 6e24e80
refactor(adaptive_throttler, resource_monitor): implement shared moni…
kxrxh 0028afc
refactor(sysmonitor): remove SetMemoryReader function from memory pla…
kxrxh 7f375ba
refactor: enhance system monitoring and adaptive throttling
kxrxh 153c015
test(sysmonitor): add comprehensive tests for Windows CPU and memory …
kxrxh 583a92b
fix(adaptive_throttler): make close public and ensure proper resource…
kxrxh d0b57ed
fix(adaptive_throttler, resource_monitor): correct loop iteration and…
kxrxh 3489be1
test: refactor adaptive throttler and monitor suites
kxrxh eb2081e
fix: stop resource monitor deterministically
kxrxh 28df064
refactor: wait for monitor shutdown with waitgroup
kxrxh f38c364
refactor(resource_monitor): simplify stop timer handling during resou…
kxrxh 1d37d65
refactor(adaptive_throttler): share rate calc, allow CPU limit 0, gua…
kxrxh fd6b529
refactor(flow): revert file renaming
kxrxh 56aa42c
docs(cpu): add thread-safety notes to CPU usage documentation
kxrxh c5c4aca
test(adaptive_throttler): enhance validation tests and moved mocks/he…
kxrxh dc4c4eb
refactor(adaptive_throttler): update config validation message
kxrxh ba8dbff
fix(flow): make AdaptiveThrottler close() private and fix channel cle…
kxrxh 0470199
fix(resource_monitor): handle panic in custom memory reader
kxrxh File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or 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 hidden or 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,121 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "math/rand" | ||
| "sync/atomic" | ||
| "time" | ||
|
|
||
| ext "github.com/reugn/go-streams/extension" | ||
| "github.com/reugn/go-streams/flow" | ||
| ) | ||
|
|
||
| func main() { | ||
| var elementsProcessed atomic.Int64 | ||
|
|
||
| // Set up demo configuration with memory simulation | ||
| throttler := setupDemoThrottler(&elementsProcessed) | ||
| defer func() { | ||
| throttler.Close() | ||
| }() | ||
|
|
||
| in := make(chan any) | ||
| out := make(chan any, 32) | ||
|
|
||
| source := ext.NewChanSource(in) | ||
| sink := ext.NewChanSink(out) | ||
|
|
||
| statsDone := make(chan struct{}) | ||
| go logThrottlerStats(throttler, statsDone) | ||
| defer close(statsDone) | ||
|
|
||
| go func() { | ||
| source. | ||
| Via(throttler). | ||
| Via(flow.NewPassThrough()). | ||
| To(sink) | ||
| }() | ||
|
|
||
| go produceBurst(in, 250) | ||
|
|
||
| for element := range sink.Out { | ||
| fmt.Printf("consumer received %v\n", element) | ||
| elementsProcessed.Add(1) // Track processed elements for memory pressure simulation | ||
| time.Sleep(25 * time.Millisecond) | ||
| } | ||
|
|
||
| fmt.Println("adaptive throttling pipeline completed") | ||
| } | ||
|
|
||
| // setupDemoThrottler creates and configures an adaptive throttler with demo settings | ||
| func setupDemoThrottler(elementsProcessed *atomic.Int64) *flow.AdaptiveThrottler { | ||
| config := flow.DefaultAdaptiveThrottlerConfig() | ||
| config.MinThroughput = 5 | ||
| config.MaxThroughput = 40 | ||
| config.SampleInterval = 200 * time.Millisecond | ||
| config.BufferSize = 32 | ||
| config.AdaptationFactor = 0.5 | ||
| config.SmoothTransitions = true | ||
| config.MaxMemoryPercent = 40.0 | ||
| config.MaxCPUPercent = 80.0 | ||
|
|
||
|
|
||
| config.MemoryReader = func() (float64, error) { | ||
| elementCount := elementsProcessed.Load() | ||
|
|
||
| // Memory pressure increases with processed elements: | ||
| // - 0-50 elements: 5% memory | ||
| // - 51-100 elements: 15% memory | ||
| // - 101-150 elements: 30% memory | ||
| // - 151+ elements: 50%+ memory (increases gradually) | ||
| var memoryPercent float64 | ||
| switch { | ||
| case elementCount <= 50: | ||
| memoryPercent = 5.0 + float64(elementCount)*0.2 // 5% to 15% | ||
| case elementCount <= 100: | ||
| memoryPercent = 15.0 + float64(elementCount-50)*0.3 // 15% to 30% | ||
| case elementCount <= 150: | ||
| memoryPercent = 30.0 + float64(elementCount-100)*0.4 // 30% to 50% | ||
| default: | ||
| memoryPercent = 50.0 + float64(elementCount-150)*0.3 // 50%+ (increases more slowly) | ||
| if memoryPercent > 95.0 { | ||
| memoryPercent = 95.0 | ||
| } | ||
| } | ||
|
|
||
| return memoryPercent, nil | ||
| } | ||
|
|
||
| return flow.NewAdaptiveThrottler(config) | ||
| } | ||
|
|
||
| func produceBurst(in chan<- any, total int) { | ||
| defer close(in) | ||
|
|
||
| for i := 0; i < total; i++ { | ||
| in <- fmt.Sprintf("job-%02d", i) | ||
|
|
||
| if (i+1)%10 == 0 { | ||
| time.Sleep(180 * time.Millisecond) | ||
| continue | ||
| } | ||
|
|
||
| time.Sleep(time.Duration(2+rand.Intn(5)) * time.Millisecond) | ||
| } | ||
| } | ||
|
|
||
| func logThrottlerStats(at *flow.AdaptiveThrottler, done <-chan struct{}) { | ||
| ticker := time.NewTicker(500 * time.Millisecond) | ||
| defer ticker.Stop() | ||
|
|
||
| for { | ||
| select { | ||
| case <-done: | ||
| return | ||
| case <-ticker.C: | ||
| stats := at.GetResourceStats() | ||
| fmt.Printf("[stats] rate=%d eps memory=%.1f%% cpu=%.1f%% goroutines=%d\n", | ||
| at.GetCurrentRate(), stats.MemoryUsedPercent, stats.CPUUsagePercent, stats.GoroutineCount) | ||
| } | ||
| } | ||
| } | ||
This file contains hidden or 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,44 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "strings" | ||
| "time" | ||
|
|
||
| ext "github.com/reugn/go-streams/extension" | ||
| "github.com/reugn/go-streams/flow" | ||
| ) | ||
|
|
||
| func editMessage(msg string) string { | ||
| return strings.ToUpper(msg) | ||
| } | ||
|
|
||
| func main() { | ||
| // To setup a custom throttler, you can modify the throttlerConfig struct with your desired values. | ||
| // For all available options, see the flow.AdaptiveThrottlerConfig struct. | ||
| throttlerConfig := flow.DefaultAdaptiveThrottlerConfig() | ||
| throttler := flow.NewAdaptiveThrottler(throttlerConfig) | ||
| defer throttler.Close() | ||
|
|
||
| in := make(chan any) | ||
|
|
||
| source := ext.NewChanSource(in) | ||
| editMapFlow := flow.NewMap(editMessage, 1) | ||
| sink := ext.NewStdoutSink() | ||
|
|
||
| go func() { | ||
| source.Via(throttler).Via(editMapFlow).To(sink) | ||
| }() | ||
|
|
||
| go func() { | ||
| defer close(in) | ||
|
|
||
| for i := 1; i <= 50; i++ { | ||
| message := fmt.Sprintf("message-%d", i) | ||
| in <- message | ||
| time.Sleep(50 * time.Millisecond) | ||
| } | ||
| }() | ||
|
|
||
| sink.AwaitCompletion() | ||
| } |
This file contains hidden or 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 hidden or 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.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.