random.go 2.6 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091
  1. /*
  2. *
  3. * Copyright 2019 gRPC authors.
  4. *
  5. * Licensed under the Apache License, Version 2.0 (the "License");
  6. * you may not use this file except in compliance with the License.
  7. * You may obtain a copy of the License at
  8. *
  9. * http://www.apache.org/licenses/LICENSE-2.0
  10. *
  11. * Unless required by applicable law or agreed to in writing, software
  12. * distributed under the License is distributed on an "AS IS" BASIS,
  13. * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14. * See the License for the specific language governing permissions and
  15. * limitations under the License.
  16. */
  17. package wrr
  18. import (
  19. "fmt"
  20. "sort"
  21. "sync"
  22. "google.golang.org/grpc/internal/grpcrand"
  23. )
  24. // weightedItem is a wrapped weighted item that is used to implement weighted random algorithm.
  25. type weightedItem struct {
  26. item interface{}
  27. weight int64
  28. accumulatedWeight int64
  29. }
  30. func (w *weightedItem) String() string {
  31. return fmt.Sprint(*w)
  32. }
  33. // randomWRR is a struct that contains weighted items implement weighted random algorithm.
  34. type randomWRR struct {
  35. mu sync.RWMutex
  36. items []*weightedItem
  37. // Are all item's weights equal
  38. equalWeights bool
  39. }
  40. // NewRandom creates a new WRR with random.
  41. func NewRandom() WRR {
  42. return &randomWRR{}
  43. }
  44. var grpcrandInt63n = grpcrand.Int63n
  45. func (rw *randomWRR) Next() (item interface{}) {
  46. rw.mu.RLock()
  47. defer rw.mu.RUnlock()
  48. if len(rw.items) == 0 {
  49. return nil
  50. }
  51. if rw.equalWeights {
  52. return rw.items[grpcrandInt63n(int64(len(rw.items)))].item
  53. }
  54. sumOfWeights := rw.items[len(rw.items)-1].accumulatedWeight
  55. // Random number in [0, sumOfWeights).
  56. randomWeight := grpcrandInt63n(sumOfWeights)
  57. // Item's accumulated weights are in ascending order, because item's weight >= 0.
  58. // Binary search rw.items to find first item whose accumulatedWeight > randomWeight
  59. // The return i is guaranteed to be in range [0, len(rw.items)) because randomWeight < last item's accumulatedWeight
  60. i := sort.Search(len(rw.items), func(i int) bool { return rw.items[i].accumulatedWeight > randomWeight })
  61. return rw.items[i].item
  62. }
  63. func (rw *randomWRR) Add(item interface{}, weight int64) {
  64. rw.mu.Lock()
  65. defer rw.mu.Unlock()
  66. accumulatedWeight := weight
  67. equalWeights := true
  68. if len(rw.items) > 0 {
  69. lastItem := rw.items[len(rw.items)-1]
  70. accumulatedWeight = lastItem.accumulatedWeight + weight
  71. equalWeights = rw.equalWeights && weight == lastItem.weight
  72. }
  73. rw.equalWeights = equalWeights
  74. rItem := &weightedItem{item: item, weight: weight, accumulatedWeight: accumulatedWeight}
  75. rw.items = append(rw.items, rItem)
  76. }
  77. func (rw *randomWRR) String() string {
  78. return fmt.Sprint(rw.items)
  79. }