-
Notifications
You must be signed in to change notification settings - Fork 156
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Remove "reset" method of ScheduledTaskRoutine to fix bugs (#369)
Signed-off-by: Daxin Wang <daxinwang@harmonycloud.cn>
- Loading branch information
Showing
7 changed files
with
158 additions
and
175 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
82 changes: 82 additions & 0 deletions
82
collector/pkg/component/analyzer/cpuanalyzer/scheduled_task.go
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,82 @@ | ||
| package cpuanalyzer | ||
|
|
||
| import ( | ||
| "errors" | ||
| "time" | ||
|
|
||
| "go.uber.org/atomic" | ||
| ) | ||
|
|
||
| type ScheduledTask interface { | ||
| run() | ||
| } | ||
|
|
||
| type ScheduledTaskRoutine struct { | ||
| expiredDuration time.Duration | ||
| ticker *time.Ticker | ||
| timer *time.Timer | ||
| stopCh chan struct{} | ||
|
|
||
| task ScheduledTask | ||
| isRunning *atomic.Bool | ||
|
|
||
| expiredCallback func() | ||
| } | ||
|
|
||
| // NewAndStartScheduledTaskRoutine creates a new routine and start it immediately. | ||
| func NewAndStartScheduledTaskRoutine( | ||
| tickerDuration time.Duration, | ||
| expiredDuration time.Duration, | ||
| task ScheduledTask, | ||
| expiredCallback func()) *ScheduledTaskRoutine { | ||
| ret := &ScheduledTaskRoutine{ | ||
| expiredDuration: expiredDuration, | ||
| ticker: time.NewTicker(tickerDuration), | ||
| timer: time.NewTimer(expiredDuration), | ||
| task: task, | ||
| isRunning: atomic.NewBool(false), | ||
| stopCh: make(chan struct{}), | ||
| expiredCallback: expiredCallback, | ||
| } | ||
| // Start the routine once it is created. | ||
| _ = ret.Start() | ||
| return ret | ||
| } | ||
|
|
||
| func (s *ScheduledTaskRoutine) Start() error { | ||
| swapped := s.isRunning.CAS(false, true) | ||
| if !swapped { | ||
| return errors.New("the routine has been started") | ||
| } | ||
| go func() { | ||
| if s.expiredCallback != nil { | ||
| defer s.expiredCallback() | ||
| } | ||
| for { | ||
| select { | ||
| case <-s.ticker.C: | ||
| // do some work | ||
| s.task.run() | ||
| case <-s.timer.C: | ||
| // The current task is expired. | ||
| s.isRunning.CAS(true, false) | ||
| s.ticker.Stop() | ||
| return | ||
| case <-s.stopCh: | ||
| s.timer.Stop() | ||
| s.ticker.Stop() | ||
| return | ||
| } | ||
| } | ||
| }() | ||
| return nil | ||
| } | ||
|
|
||
| func (s *ScheduledTaskRoutine) Stop() error { | ||
| swapped := s.isRunning.CAS(true, false) | ||
| if !swapped { | ||
| return errors.New("the routine is not running") | ||
| } | ||
| s.stopCh <- struct{}{} | ||
| return nil | ||
| } |
36 changes: 36 additions & 0 deletions
36
collector/pkg/component/analyzer/cpuanalyzer/scheduled_task_test.go
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,36 @@ | ||
| package cpuanalyzer | ||
|
|
||
| import ( | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/stretchr/testify/assert" | ||
| ) | ||
|
|
||
| func TestScheduledTask(t *testing.T) { | ||
| // test case 1: Normal expired exit | ||
| task1 := &testIncrementTask{0} | ||
| routine1 := NewAndStartScheduledTaskRoutine(1*time.Millisecond, 5*time.Millisecond, task1, nil) | ||
| _ = routine1.Start() | ||
| time.Sleep(10 * time.Millisecond) | ||
| assert.Equal(t, false, routine1.isRunning.Load()) | ||
| assert.Equal(t, 5, task1.count) | ||
|
|
||
| // Case 2: Double start or double stop | ||
| task3 := &testIncrementTask{0} | ||
| routine3 := NewAndStartScheduledTaskRoutine(1*time.Millisecond, 5*time.Millisecond, task3, nil) | ||
| err := routine3.Start() | ||
| assert.Error(t, err) | ||
| err = routine3.Stop() | ||
| assert.NoError(t, err) | ||
| err = routine3.Stop() | ||
| assert.Error(t, err) | ||
| } | ||
|
|
||
| type testIncrementTask struct { | ||
| count int | ||
| } | ||
|
|
||
| func (t *testIncrementTask) run() { | ||
| t.count++ | ||
| } |
24 changes: 24 additions & 0 deletions
24
collector/pkg/component/analyzer/cpuanalyzer/self_metrics.go
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,24 @@ | ||
| package cpuanalyzer | ||
|
|
||
| import ( | ||
| "context" | ||
| "sync" | ||
|
|
||
| "go.opentelemetry.io/otel/metric" | ||
| ) | ||
|
|
||
| var onceMetric sync.Once | ||
|
|
||
| const ( | ||
| goroutineSize = "kindling_telemetry_cpuanalyzer_routine_size" | ||
| ) | ||
|
|
||
| func newSelfMetrics(meterProvider metric.MeterProvider, analyzer *CpuAnalyzer) { | ||
| onceMetric.Do(func() { | ||
| meter := metric.Must(meterProvider.Meter("kindling")) | ||
| meter.NewInt64GaugeObserver(goroutineSize, | ||
| func(ctx context.Context, result metric.Int64ObserverResult) { | ||
| result.Observe(int64(analyzer.routineSize.Load())) | ||
| }) | ||
| }) | ||
| } |
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.