loop.go 2.8 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495
  1. /*
  2. Copyright 2023 The Kubernetes Authors.
  3. Licensed under the Apache License, Version 2.0 (the "License");
  4. you may not use this file except in compliance with the License.
  5. You may obtain a copy of the License at
  6. http://www.apache.org/licenses/LICENSE-2.0
  7. Unless required by applicable law or agreed to in writing, software
  8. distributed under the License is distributed on an "AS IS" BASIS,
  9. WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  10. See the License for the specific language governing permissions and
  11. limitations under the License.
  12. */
  13. package wait
  14. import (
  15. "context"
  16. "time"
  17. "k8s.io/apimachinery/pkg/util/runtime"
  18. )
  19. // loopConditionUntilContext executes the provided condition at intervals defined by
  20. // the provided timer until the provided context is cancelled, the condition returns
  21. // true, or the condition returns an error. If sliding is true, the period is computed
  22. // after condition runs. If it is false then period includes the runtime for condition.
  23. // If immediate is false the first delay happens before any call to condition, if
  24. // immediate is true the condition will be invoked before waiting and guarantees that
  25. // the condition is invoked at least once, regardless of whether the context has been
  26. // cancelled. The returned error is the error returned by the last condition or the
  27. // context error if the context was terminated.
  28. //
  29. // This is the common loop construct for all polling in the wait package.
  30. func loopConditionUntilContext(ctx context.Context, t Timer, immediate, sliding bool, condition ConditionWithContextFunc) error {
  31. defer t.Stop()
  32. var timeCh <-chan time.Time
  33. doneCh := ctx.Done()
  34. if !sliding {
  35. timeCh = t.C()
  36. }
  37. // if immediate is true the condition is
  38. // guaranteed to be executed at least once,
  39. // if we haven't requested immediate execution, delay once
  40. if immediate {
  41. if ok, err := func() (bool, error) {
  42. defer runtime.HandleCrashWithContext(ctx)
  43. return condition(ctx)
  44. }(); err != nil || ok {
  45. return err
  46. }
  47. }
  48. if sliding {
  49. timeCh = t.C()
  50. }
  51. for {
  52. // Wait for either the context to be cancelled or the next invocation be called
  53. select {
  54. case <-doneCh:
  55. return ctx.Err()
  56. case <-timeCh:
  57. }
  58. // IMPORTANT: Because there is no channel priority selection in golang
  59. // it is possible for very short timers to "win" the race in the previous select
  60. // repeatedly even when the context has been canceled. We therefore must
  61. // explicitly check for context cancellation on every loop and exit if true to
  62. // guarantee that we don't invoke condition more than once after context has
  63. // been cancelled.
  64. if err := ctx.Err(); err != nil {
  65. return err
  66. }
  67. if !sliding {
  68. t.Next()
  69. }
  70. if ok, err := func() (bool, error) {
  71. defer runtime.HandleCrashWithContext(ctx)
  72. return condition(ctx)
  73. }(); err != nil || ok {
  74. return err
  75. }
  76. if sliding {
  77. t.Next()
  78. }
  79. }
  80. }