stress_composite.go 2.0 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889
  1. // Copyright 2018 The etcd Authors
  2. //
  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. //
  7. // http://www.apache.org/licenses/LICENSE-2.0
  8. //
  9. // Unless required by applicable law or agreed to in writing, software
  10. // distributed under the License is distributed on an "AS IS" BASIS,
  11. // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. // See the License for the specific language governing permissions and
  13. // limitations under the License.
  14. package tester
  15. import "sync"
  16. // compositeStresser implements a Stresser that runs a slice of
  17. // stressing clients concurrently.
  18. type compositeStresser struct {
  19. stressers []Stresser
  20. }
  21. func (cs *compositeStresser) Stress() error {
  22. for i, s := range cs.stressers {
  23. if err := s.Stress(); err != nil {
  24. for j := 0; j < i; j++ {
  25. cs.stressers[j].Close()
  26. }
  27. return err
  28. }
  29. }
  30. return nil
  31. }
  32. func (cs *compositeStresser) Pause() (ems map[string]int) {
  33. ems = make(map[string]int)
  34. var wg sync.WaitGroup
  35. wg.Add(len(cs.stressers))
  36. for i := range cs.stressers {
  37. go func(s Stresser) {
  38. defer wg.Done()
  39. errs := s.Pause()
  40. for k, v := range errs {
  41. ems[k] += v
  42. }
  43. }(cs.stressers[i])
  44. }
  45. wg.Wait()
  46. return ems
  47. }
  48. func (cs *compositeStresser) Close() (ems map[string]int) {
  49. ems = make(map[string]int)
  50. var wg sync.WaitGroup
  51. wg.Add(len(cs.stressers))
  52. for i := range cs.stressers {
  53. go func(s Stresser) {
  54. defer wg.Done()
  55. errs := s.Close()
  56. for k, v := range errs {
  57. ems[k] += v
  58. }
  59. }(cs.stressers[i])
  60. }
  61. wg.Wait()
  62. return ems
  63. }
  64. func (cs *compositeStresser) ModifiedKeys() (modifiedKey int64) {
  65. for _, stress := range cs.stressers {
  66. modifiedKey += stress.ModifiedKeys()
  67. }
  68. return modifiedKey
  69. }
  70. func (cs *compositeStresser) Checker() Checker {
  71. var chks []Checker
  72. for _, s := range cs.stressers {
  73. if chk := s.Checker(); chk != nil {
  74. chks = append(chks, chk)
  75. }
  76. }
  77. if len(chks) == 0 {
  78. return nil
  79. }
  80. return newCompositeChecker(chks)
  81. }