Commit f6be1b46e4fe1313bdc6aa5d97c9d8444d62cb4a - golang-github-hashicorp-serf

Adds adjustment factor back in and fixes random phantom with non-zero mean. James Phillips 8 years ago

6 changed file(s) with 96 addition(s) and 38 deletion(s). Raw diff Collapse all Expand all

+52

-11

coordinate/client.go less more

18	18	// the algorithm.
19	19	config *Config
20	20
	21	// adjustmentIndex is the current index into the adjustmentSamples slice.
	22	adjustmentIndex uint
	23
	24	// adjustment is used to store samples for the adjustment calculation.
	25	adjustmentSamples []float64
	26
21	27	// mutex enables safe concurrent access to the client.
22	28	mutex *sync.RWMutex
23	29	}

36	42	return &Client{
37	43	coord: NewCoordinate(config),
38	44	config: config,
	45	adjustmentIndex: 0,
	46	adjustmentSamples: make([]float64, config.AdjustmentWindowSize),
39	47	mutex: &sync.RWMutex{},
40	48	}, nil
41	49	}

48	56	return c.coord.Clone()
49	57	}
50	58
51		// Update takes other, a coordinate for another node, and rtt, a round trip
52		// time observation for a ping to that node, and updates the estimated position of
53		// the client's coordinate.
54		func (c Client) Update(other Coordinate, rtt time.Duration) {
55		c.mutex.Lock()
56		defer c.mutex.Unlock()
57
	59	// updateVivialdi updates the Vivaldi portion of the client's coordinate. This
	60	// assumes that the mutex has been locked already.
	61	func (c Client) updateVivaldi(other Coordinate, rttSeconds float64) {
58	62	const zeroThreshold = 1.0e-6
59	63
60	64	dist := c.coord.DistanceTo(other)
61		rttSeconds := rtt.Seconds()
62	65	if rttSeconds < zeroThreshold {
63	66	rttSeconds = zeroThreshold
64	67	}

70	73	}
71	74	weight := c.coord.Error / totalError
72	75
73		c.coord.Error = c.config.VivaldiCEweightwrongness + c.coord.Error(1-c.config.VivaldiCEweight)
	76	c.coord.Error = c.config.VivaldiCEweightwrongness + c.coord.Error(1.0-c.config.VivaldiCEweight)
74	77	if c.coord.Error > c.config.VivaldiErrorMax {
75	78	c.coord.Error = c.config.VivaldiErrorMax
76	79	}

80	83	c.coord = c.coord.ApplyForce(force, other)
81	84	}
82	85
	86	// updateAdjustment updates the adjustment portion of the client's coordinate, if
	87	// the feature is enabled. This assumes that the mutex has been locked already.
	88	func (c Client) updateAdjustment(other Coordinate, rttSeconds float64) {
	89	if c.config.AdjustmentWindowSize == 0 {
	90	return
	91	}
	92
	93	dist := c.coord.DistanceTo(other)
	94	c.adjustmentSamples[c.adjustmentIndex] = rttSeconds - dist
	95	c.adjustmentIndex = (c.adjustmentIndex + 1) % c.config.AdjustmentWindowSize
	96
	97	sum := 0.0
	98	for _, sample := range c.adjustmentSamples {
	99	sum += sample
	100	}
	101	c.coord.Adjustment = sum / (2.0*float64(c.config.AdjustmentWindowSize))
	102	}
	103
	104	// Update takes other, a coordinate for another node, and rtt, a round trip
	105	// time observation for a ping to that node, and updates the estimated position of
	106	// the client's coordinate.
	107	func (c Client) Update(other Coordinate, rtt time.Duration) {
	108	c.mutex.Lock()
	109	defer c.mutex.Unlock()
	110
	111	rttSeconds := rtt.Seconds()
	112	c.updateVivaldi(other, rttSeconds)
	113	c.updateAdjustment(other, rttSeconds)
	114	}
	115
83	116	// DistanceTo returns the estimated RTT from the client's coordinate to other, the
84	117	// coordinate for another node.
85	118	func (c Client) DistanceTo(other Coordinate) time.Duration {
86	119	c.mutex.RLock()
87	120	defer c.mutex.RUnlock()
88	121
89		dist := c.coord.DistanceTo(other) * secondsToNanoseconds
90		return time.Duration(dist)
	122	// It's important that the adjustment values are summed here, and not down
	123	// in the coordinate's DistanceTo() function, because the calculation of
	124	// the adjustment is based only on the current Vivaldi distance, and not
	125	// the current adjustment factors.
	126	dist := c.coord.DistanceTo(other)
	127	adjustedDist := dist + c.coord.Adjustment + other.Adjustment
	128	if adjustedDist > 0.0 {
	129	dist = adjustedDist
	130	}
	131	return time.Duration(dist*secondsToNanoseconds)
91	132	}

+17

-2

coordinate/client_test.go less more

44	44	// client expects, given its distance.
45	45	other := NewCoordinate(config)
46	46	other.Vec[2] = 0.001
47		rtt := time.Duration(2.0 * other.Vec[2] * secondsToNanoseconds)
	47	rtt := time.Duration(2.0other.Vec[2]secondsToNanoseconds)
48	48	client.Update(other, rtt)
49	49
50	50	// The client should have scooted down to get away from it.

66	66	// Fiddle a raw coordinate to put it a specific number of seconds away.
67	67	other := NewCoordinate(config)
68	68	other.Vec[2] = 12.345
69		expected := time.Duration(other.Vec[2] * secondsToNanoseconds)
	69	expected := time.Duration(other.Vec[2]*secondsToNanoseconds)
70	70	dist := client.DistanceTo(other)
71	71	if dist != expected {
72	72	t.Fatalf("distance doesn't match %9.6f != %9.6f", dist.Seconds(), expected.Seconds())
73	73	}
	74
	75	// Make sure negative adjustment factors are ignored.
	76	client.coord.Adjustment = -(other.Vec[2] + 0.1)
	77	dist = client.DistanceTo(other)
	78	if dist != expected {
	79	t.Fatalf("distance doesn't match %9.6f != %9.6f", dist.Seconds(), expected.Seconds())
	80	}
	81
	82	// Make sure positive adjustment factors affect the distance.
	83	client.coord.Adjustment = 0.1
	84	expected = time.Duration((other.Vec[2] + 0.1)*secondsToNanoseconds)
	85	dist = client.DistanceTo(other)
	86	if dist != expected {
	87	t.Fatalf("distance doesn't match %9.6f != %9.6f", dist.Seconds(), expected.Seconds())
	88	}
74	89	}

+10

-4

coordinate/config.go less more

30	30	// VivaldiCC is a tuning factor that controls the maximum impact an
31	31	// observation can have on a node's coordinate. See [1] for more details.
32	32	VivaldiCC float64
	33
	34	// AdjustmentWindowSize is a tuning factor that determines how many samples
	35	// we retain to calculate the adjustment factor as discussed in [3]. Setting
	36	// this to zero disables this feature.
	37	AdjustmentWindowSize uint
33	38	}
34	39
35	40	// DefaultConfig returns a Config that has some default values suitable for
36	41	// basic testing of the algorithm, but not tuned to any particular type of cluster.
37	42	func DefaultConfig() *Config {
38	43	return &Config{
39		Dimensionality: 8,
40		VivaldiErrorMax: 1.5,
41		VivaldiCE: 0.25,
42		VivaldiCC: 0.25,
	44	Dimensionality: 8,
	45	VivaldiErrorMax: 1.5,
	46	VivaldiCE: 0.25,
	47	VivaldiCC: 0.25,
	48	AdjustmentWindowSize: 20,
43	49	}
44	50	}

-3

coordinate/coordinate.go less more

69	69	panic(ErrDimensionalityConflict)
70	70	}
71	71
72		euclidianPart := magnitude(diff(c.Vec, other.Vec))
73		adjustmentPart := c.Adjustment + other.Adjustment
74		return euclidianPart + adjustmentPart
	72	return magnitude(diff(c.Vec, other.Vec))
75	73	}
76	74
77	75	// add returns the sum of vec1 and vec2. This assumes the dimensions have

+10

-13

coordinate/performance_test.go less more

15	15	truth := GenerateLine(nodes, spacing)
16	16	Simulate(clients, truth, cycles, nil)
17	17	stats := Evaluate(clients, truth)
18		if stats.ErrorAvg > 0.005 {
19		t.Fatalf("average error is too large, %9.6f", stats.ErrorAvg)
	18	if stats.ErrorAvg > 0.004 \|\| stats.ErrorMax > 0.015 {
	19	t.Fatalf("performance stats are out of spec: %v", stats)
20	20	}
21	21	}
22	22

31	31	truth := GenerateGrid(nodes, spacing)
32	32	Simulate(clients, truth, cycles, nil)
33	33	stats := Evaluate(clients, truth)
34		if stats.ErrorAvg > 0.006 {
35		t.Fatalf("average error is too large, %9.6f", stats.ErrorAvg)
	34	if stats.ErrorAvg > 0.005 \|\| stats.ErrorMax > 0.051 {
	35	t.Fatalf("performance stats are out of spec: %v", stats)
36	36	}
37	37	}
38	38

47	47	truth := GenerateSplit(nodes, lan, wan)
48	48	Simulate(clients, truth, cycles, nil)
49	49	stats := Evaluate(clients, truth)
50		if stats.ErrorAvg > 0.045 {
51		t.Fatalf("average error is too large, %9.6f", stats.ErrorAvg)
	50	if stats.ErrorAvg > 0.044 \|\| stats.ErrorMax > 0.343 {
	51	t.Fatalf("performance stats are out of spec: %v", stats)
52	52	}
53	53	}
54	54
55	55	func TestPerformance_Random(t *testing.T) {
56		const max = 10*time.Millisecond
	56	const mean, deviation = 100time.Millisecond, 10time.Millisecond
57	57	const nodes, cycles = 25, 1000
58	58	config := DefaultConfig()
59	59	clients, err := GenerateClients(nodes, config)
60	60	if err != nil {
61	61	t.Fatal(err)
62	62	}
63		truth := GenerateRandom(nodes, max)
	63	truth := GenerateRandom(nodes, mean, deviation)
64	64	Simulate(clients, truth, cycles, nil)
65	65	stats := Evaluate(clients, truth)
66
67		// TODO - Currently horrible! Height and the adjustment factor should
68		// help here, so revisit once those are in.
69		if stats.ErrorAvg > 4.8 {
70		t.Fatalf("average error is too large, %9.6f", stats.ErrorAvg)
	66	if stats.ErrorAvg > 0.079 \|\| stats.ErrorMax > 0.363 {
	67	t.Fatalf("performance stats are out of spec: %v", stats)
71	68	}
72	69	}

-5

coordinate/phantom.go less more

83	83	return truth
84	84	}
85	85
86		// GenerateRandom returns a truth matrix for a set of nodes with random delays, up
87		// to the given max. The RNG is re-seeded so you always get the same matrix for a
88		// given size.
89		func GenerateRandom(nodes int, max time.Duration) [][]time.Duration {
	86	// GenerateRandom returns a truth matrix for a set of nodes with normally
	87	// distributed delays, with the given mean and deviation. The RNG is re-seeded
	88	// so you always get the same matrix for a given size.
	89	func GenerateRandom(nodes int, mean time.Duration, deviation time.Duration) [][]time.Duration {
90	90	rand.Seed(1)
91	91
92	92	truth := make([][]time.Duration, nodes)

96	96
97	97	for i := 0; i < nodes; i++ {
98	98	for j := i + 1; j < nodes; j++ {
99		rtt := time.Duration(rand.Float64() * float64(max))
	99	rttSeconds := rand.NormFloat64() * deviation.Seconds() + mean.Seconds()
	100	rtt := time.Duration(rttSeconds * secondsToNanoseconds)
100	101	truth[i][j], truth[j][i] = rtt, rtt
101	102	}
102	103	}