1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220
|
package hostpool
import (
"log"
"math/rand"
"time"
)
type epsilonHostPoolResponse struct {
standardHostPoolResponse
started time.Time
ended time.Time
}
func (r *epsilonHostPoolResponse) Mark(err error) {
r.Do(func() {
r.ended = time.Now()
doMark(err, r)
})
}
type epsilonGreedyHostPool struct {
standardHostPool // TODO - would be nifty if we could embed HostPool and Locker interfaces
epsilon float32 // this is our exploration factor
decayDuration time.Duration
EpsilonValueCalculator // embed the epsilonValueCalculator
timer
quit chan bool
}
// Construct an Epsilon Greedy HostPool
//
// Epsilon Greedy is an algorithm that allows HostPool not only to track failure state,
// but also to learn about "better" options in terms of speed, and to pick from available hosts
// based on how well they perform. This gives a weighted request rate to better
// performing hosts, while still distributing requests to all hosts (proportionate to their performance).
// The interface is the same as the standard HostPool, but be sure to mark the HostResponse immediately
// after executing the request to the host, as that will stop the implicitly running request timer.
//
// A good overview of Epsilon Greedy is here http://stevehanov.ca/blog/index.php?id=132
//
// To compute the weighting scores, we perform a weighted average of recent response times, over the course of
// `decayDuration`. decayDuration may be set to 0 to use the default value of 5 minutes
// We then use the supplied EpsilonValueCalculator to calculate a score from that weighted average response time.
func NewEpsilonGreedy(hosts []string, decayDuration time.Duration, calc EpsilonValueCalculator) HostPool {
if decayDuration <= 0 {
decayDuration = defaultDecayDuration
}
stdHP := New(hosts).(*standardHostPool)
p := &epsilonGreedyHostPool{
standardHostPool: *stdHP,
epsilon: float32(initialEpsilon),
decayDuration: decayDuration,
EpsilonValueCalculator: calc,
timer: &realTimer{},
quit: make(chan bool),
}
// allocate structures
for _, h := range p.hostList {
h.epsilonCounts = make([]int64, epsilonBuckets)
h.epsilonValues = make([]int64, epsilonBuckets)
}
go p.epsilonGreedyDecay()
return p
}
func (p *epsilonGreedyHostPool) Close() {
// No need to do p.quit <- true as close(p.quit) does the trick.
close(p.quit)
}
func (p *epsilonGreedyHostPool) SetEpsilon(newEpsilon float32) {
p.Lock()
defer p.Unlock()
p.epsilon = newEpsilon
}
func (p *epsilonGreedyHostPool) SetHosts(hosts []string) {
p.Lock()
defer p.Unlock()
p.standardHostPool.setHosts(hosts)
for _, h := range p.hostList {
h.epsilonCounts = make([]int64, epsilonBuckets)
h.epsilonValues = make([]int64, epsilonBuckets)
}
}
func (p *epsilonGreedyHostPool) epsilonGreedyDecay() {
durationPerBucket := p.decayDuration / epsilonBuckets
ticker := time.NewTicker(durationPerBucket)
for {
select {
case <-p.quit:
ticker.Stop()
return
case <-ticker.C:
p.performEpsilonGreedyDecay()
}
}
}
func (p *epsilonGreedyHostPool) performEpsilonGreedyDecay() {
p.Lock()
for _, h := range p.hostList {
h.epsilonIndex += 1
h.epsilonIndex = h.epsilonIndex % epsilonBuckets
h.epsilonCounts[h.epsilonIndex] = 0
h.epsilonValues[h.epsilonIndex] = 0
}
p.Unlock()
}
func (p *epsilonGreedyHostPool) Get() HostPoolResponse {
p.Lock()
defer p.Unlock()
host := p.getEpsilonGreedy()
if host == "" {
return nil
}
started := time.Now()
return &epsilonHostPoolResponse{
standardHostPoolResponse: standardHostPoolResponse{host: host, pool: p},
started: started,
}
}
func (p *epsilonGreedyHostPool) getEpsilonGreedy() string {
var hostToUse *hostEntry
// this is our exploration phase
if rand.Float32() < p.epsilon {
p.epsilon = p.epsilon * epsilonDecay
if p.epsilon < minEpsilon {
p.epsilon = minEpsilon
}
return p.getRoundRobin()
}
// calculate values for each host in the 0..1 range (but not ormalized)
var possibleHosts []*hostEntry
now := time.Now()
var sumValues float64
for _, h := range p.hostList {
if h.canTryHost(now) {
v := h.getWeightedAverageResponseTime()
if v > 0 {
ev := p.CalcValueFromAvgResponseTime(v)
h.epsilonValue = ev
sumValues += ev
possibleHosts = append(possibleHosts, h)
}
}
}
if len(possibleHosts) != 0 {
// now normalize to the 0..1 range to get a percentage
for _, h := range possibleHosts {
h.epsilonPercentage = h.epsilonValue / sumValues
}
// do a weighted random choice among hosts
ceiling := 0.0
pickPercentage := rand.Float64()
for _, h := range possibleHosts {
ceiling += h.epsilonPercentage
if pickPercentage <= ceiling {
hostToUse = h
break
}
}
}
if hostToUse == nil {
if len(possibleHosts) != 0 {
log.Println("Failed to randomly choose a host, Dan loses")
}
return p.getRoundRobin()
}
if hostToUse.dead {
hostToUse.willRetryHost(p.maxRetryInterval)
}
return hostToUse.host
}
func (p *epsilonGreedyHostPool) markSuccess(hostR HostPoolResponse) {
// first do the base markSuccess - a little redundant with host lookup but cleaner than repeating logic
p.standardHostPool.markSuccess(hostR)
eHostR, ok := hostR.(*epsilonHostPoolResponse)
if !ok {
log.Printf("Incorrect type in eps markSuccess!") // TODO reflection to print out offending type
return
}
host := eHostR.host
duration := p.between(eHostR.started, eHostR.ended)
p.Lock()
defer p.Unlock()
h, ok := p.hosts[host]
if !ok {
log.Fatalf("host %s not in HostPool %v", host, p.Hosts())
}
h.epsilonCounts[h.epsilonIndex]++
h.epsilonValues[h.epsilonIndex] += int64(duration.Seconds() * 1000)
}
// --- timer: this just exists for testing
type timer interface {
between(time.Time, time.Time) time.Duration
}
type realTimer struct{}
func (rt *realTimer) between(start time.Time, end time.Time) time.Duration {
return end.Sub(start)
}
|