mirror of
https://github.com/tendermint/tendermint.git
synced 2026-01-06 13:26:23 +00:00
Merge pull request #917 from tendermint/trust-metric-cleanup
Trust metric cleanup
This commit is contained in:
Ethan Buchman
GitHub
56
p2p/trust/config.go
Normal file
56
p2p/trust/config.go
Normal file
@@ -0,0 +1,56 @@
|
||||
package trust
|
||||
|
||||
import "time"
|
||||
|
||||
// TrustMetricConfig - Configures the weight functions and time intervals for the metric
|
||||
type TrustMetricConfig struct {
|
||||
// Determines the percentage given to current behavior
|
||||
ProportionalWeight float64
|
||||
|
||||
// Determines the percentage given to prior behavior
|
||||
IntegralWeight float64
|
||||
|
||||
// The window of time that the trust metric will track events across.
|
||||
// This can be set to cover many days without issue
|
||||
TrackingWindow time.Duration
|
||||
|
||||
// Each interval should be short for adapability.
|
||||
// Less than 30 seconds is too sensitive,
|
||||
// and greater than 5 minutes will make the metric numb
|
||||
IntervalLength time.Duration
|
||||
}
|
||||
|
||||
// DefaultConfig returns a config with values that have been tested and produce desirable results
|
||||
func DefaultConfig() TrustMetricConfig {
|
||||
return TrustMetricConfig{
|
||||
ProportionalWeight: 0.4,
|
||||
IntegralWeight: 0.6,
|
||||
TrackingWindow: (time.Minute * 60 * 24) * 14, // 14 days.
|
||||
IntervalLength: 1 * time.Minute,
|
||||
}
|
||||
}
|
||||
|
||||
// Ensures that all configuration elements have valid values
|
||||
func customConfig(tmc TrustMetricConfig) TrustMetricConfig {
|
||||
config := DefaultConfig()
|
||||
|
||||
// Check the config for set values, and setup appropriately
|
||||
if tmc.ProportionalWeight > 0 {
|
||||
config.ProportionalWeight = tmc.ProportionalWeight
|
||||
}
|
||||
|
||||
if tmc.IntegralWeight > 0 {
|
||||
config.IntegralWeight = tmc.IntegralWeight
|
||||
}
|
||||
|
||||
if tmc.IntervalLength > time.Duration(0) {
|
||||
config.IntervalLength = tmc.IntervalLength
|
||||
}
|
||||
|
||||
if tmc.TrackingWindow > time.Duration(0) &&
|
||||
tmc.TrackingWindow >= config.IntervalLength {
|
||||
config.TrackingWindow = tmc.TrackingWindow
|
||||
}
|
||||
|
||||
return config
|
||||
}
|
||||
@@ -4,194 +4,11 @@
|
||||
package trust
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"math"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
cmn "github.com/tendermint/tmlibs/common"
|
||||
dbm "github.com/tendermint/tmlibs/db"
|
||||
)
|
||||
|
||||
const defaultStorePeriodicSaveInterval = 1 * time.Minute
|
||||
|
||||
// TrustMetricStore - Manages all trust metrics for peers
|
||||
type TrustMetricStore struct {
|
||||
cmn.BaseService
|
||||
|
||||
// Maps a Peer.Key to that peer's TrustMetric
|
||||
peerMetrics map[string]*TrustMetric
|
||||
|
||||
// Mutex that protects the map and history data file
|
||||
mtx sync.Mutex
|
||||
|
||||
// The db where peer trust metric history data will be stored
|
||||
db dbm.DB
|
||||
|
||||
// This configuration will be used when creating new TrustMetrics
|
||||
config TrustMetricConfig
|
||||
}
|
||||
|
||||
// NewTrustMetricStore returns a store that saves data to the DB
|
||||
// and uses the config when creating new trust metrics
|
||||
func NewTrustMetricStore(db dbm.DB, tmc TrustMetricConfig) *TrustMetricStore {
|
||||
tms := &TrustMetricStore{
|
||||
peerMetrics: make(map[string]*TrustMetric),
|
||||
db: db,
|
||||
config: tmc,
|
||||
}
|
||||
|
||||
tms.BaseService = *cmn.NewBaseService(nil, "TrustMetricStore", tms)
|
||||
return tms
|
||||
}
|
||||
|
||||
// OnStart implements Service
|
||||
func (tms *TrustMetricStore) OnStart() error {
|
||||
if err := tms.BaseService.OnStart(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
tms.mtx.Lock()
|
||||
defer tms.mtx.Unlock()
|
||||
|
||||
tms.loadFromDB()
|
||||
go tms.saveRoutine()
|
||||
return nil
|
||||
}
|
||||
|
||||
// OnStop implements Service
|
||||
func (tms *TrustMetricStore) OnStop() {
|
||||
tms.BaseService.OnStop()
|
||||
|
||||
tms.mtx.Lock()
|
||||
defer tms.mtx.Unlock()
|
||||
|
||||
// Stop all trust metric go-routines
|
||||
for _, tm := range tms.peerMetrics {
|
||||
tm.Stop()
|
||||
}
|
||||
|
||||
// Make the final trust history data save
|
||||
tms.saveToDB()
|
||||
}
|
||||
|
||||
// Size returns the number of entries in the trust metric store
|
||||
func (tms *TrustMetricStore) Size() int {
|
||||
tms.mtx.Lock()
|
||||
defer tms.mtx.Unlock()
|
||||
|
||||
return tms.size()
|
||||
}
|
||||
|
||||
// GetPeerTrustMetric returns a trust metric by peer key
|
||||
func (tms *TrustMetricStore) GetPeerTrustMetric(key string) *TrustMetric {
|
||||
tms.mtx.Lock()
|
||||
defer tms.mtx.Unlock()
|
||||
|
||||
tm, ok := tms.peerMetrics[key]
|
||||
if !ok {
|
||||
// If the metric is not available, we will create it
|
||||
tm = NewMetricWithConfig(tms.config)
|
||||
// The metric needs to be in the map
|
||||
tms.peerMetrics[key] = tm
|
||||
}
|
||||
return tm
|
||||
}
|
||||
|
||||
// PeerDisconnected pauses the trust metric associated with the peer identified by the key
|
||||
func (tms *TrustMetricStore) PeerDisconnected(key string) {
|
||||
tms.mtx.Lock()
|
||||
defer tms.mtx.Unlock()
|
||||
|
||||
// If the Peer that disconnected has a metric, pause it
|
||||
if tm, ok := tms.peerMetrics[key]; ok {
|
||||
tm.Pause()
|
||||
}
|
||||
}
|
||||
|
||||
// Saves the history data for all peers to the store DB.
|
||||
// This public method acquires the trust metric store lock
|
||||
func (tms *TrustMetricStore) SaveToDB() {
|
||||
tms.mtx.Lock()
|
||||
defer tms.mtx.Unlock()
|
||||
|
||||
tms.saveToDB()
|
||||
}
|
||||
|
||||
/* Private methods */
|
||||
|
||||
// size returns the number of entries in the store without acquiring the mutex
|
||||
func (tms *TrustMetricStore) size() int {
|
||||
return len(tms.peerMetrics)
|
||||
}
|
||||
|
||||
/* Loading & Saving */
|
||||
/* Both loadFromDB and savetoDB assume the mutex has been acquired */
|
||||
|
||||
var trustMetricKey = []byte("trustMetricStore")
|
||||
|
||||
// Loads the history data for all peers from the store DB
|
||||
// cmn.Panics if file is corrupt
|
||||
func (tms *TrustMetricStore) loadFromDB() bool {
|
||||
// Obtain the history data we have so far
|
||||
bytes := tms.db.Get(trustMetricKey)
|
||||
if bytes == nil {
|
||||
return false
|
||||
}
|
||||
|
||||
peers := make(map[string]MetricHistoryJSON, 0)
|
||||
err := json.Unmarshal(bytes, &peers)
|
||||
if err != nil {
|
||||
cmn.PanicCrisis(cmn.Fmt("Could not unmarshal Trust Metric Store DB data: %v", err))
|
||||
}
|
||||
|
||||
// If history data exists in the file,
|
||||
// load it into trust metric
|
||||
for key, p := range peers {
|
||||
tm := NewMetricWithConfig(tms.config)
|
||||
|
||||
tm.Init(p)
|
||||
// Load the peer trust metric into the store
|
||||
tms.peerMetrics[key] = tm
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// Saves the history data for all peers to the store DB
|
||||
func (tms *TrustMetricStore) saveToDB() {
|
||||
tms.Logger.Debug("Saving TrustHistory to DB", "size", tms.size())
|
||||
|
||||
peers := make(map[string]MetricHistoryJSON, 0)
|
||||
|
||||
for key, tm := range tms.peerMetrics {
|
||||
// Add an entry for the peer identified by key
|
||||
peers[key] = tm.HistoryJSON()
|
||||
}
|
||||
|
||||
// Write all the data back to the DB
|
||||
bytes, err := json.Marshal(peers)
|
||||
if err != nil {
|
||||
tms.Logger.Error("Failed to encode the TrustHistory", "err", err)
|
||||
return
|
||||
}
|
||||
tms.db.SetSync(trustMetricKey, bytes)
|
||||
}
|
||||
|
||||
// Periodically saves the trust history data to the DB
|
||||
func (tms *TrustMetricStore) saveRoutine() {
|
||||
t := time.NewTicker(defaultStorePeriodicSaveInterval)
|
||||
defer t.Stop()
|
||||
loop:
|
||||
for {
|
||||
select {
|
||||
case <-t.C:
|
||||
tms.SaveToDB()
|
||||
case <-tms.Quit:
|
||||
break loop
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
//---------------------------------------------------------------------------------------
|
||||
|
||||
const (
|
||||
@@ -205,6 +22,12 @@ const (
|
||||
defaultHistoryDataWeight = 0.8
|
||||
)
|
||||
|
||||
// MetricHistoryJSON - history data necessary to save the trust metric
|
||||
type MetricHistoryJSON struct {
|
||||
NumIntervals int `json:"intervals"`
|
||||
History []float64 `json:"history"`
|
||||
}
|
||||
|
||||
// TrustMetric - keeps track of peer reliability
|
||||
// See tendermint/docs/architecture/adr-006-trust-metric.md for details
|
||||
type TrustMetric struct {
|
||||
@@ -254,10 +77,31 @@ type TrustMetric struct {
|
||||
stop chan struct{}
|
||||
}
|
||||
|
||||
// MetricHistoryJSON - history data necessary to save the trust metric
|
||||
type MetricHistoryJSON struct {
|
||||
NumIntervals int `json:"intervals"`
|
||||
History []float64 `json:"history"`
|
||||
// NewMetric returns a trust metric with the default configuration
|
||||
func NewMetric() *TrustMetric {
|
||||
return NewMetricWithConfig(DefaultConfig())
|
||||
}
|
||||
|
||||
// NewMetricWithConfig returns a trust metric with a custom configuration
|
||||
func NewMetricWithConfig(tmc TrustMetricConfig) *TrustMetric {
|
||||
tm := new(TrustMetric)
|
||||
config := customConfig(tmc)
|
||||
|
||||
// Setup using the configuration values
|
||||
tm.proportionalWeight = config.ProportionalWeight
|
||||
tm.integralWeight = config.IntegralWeight
|
||||
tm.intervalLen = config.IntervalLength
|
||||
// The maximum number of time intervals is the tracking window / interval length
|
||||
tm.maxIntervals = int(config.TrackingWindow / tm.intervalLen)
|
||||
// The history size will be determined by the maximum number of time intervals
|
||||
tm.historyMaxSize = intervalToHistoryOffset(tm.maxIntervals) + 1
|
||||
// This metric has a perfect history so far
|
||||
tm.historyValue = 1.0
|
||||
// Setup the stop channel
|
||||
tm.stop = make(chan struct{})
|
||||
|
||||
go tm.processRequests()
|
||||
return tm
|
||||
}
|
||||
|
||||
// Returns a snapshot of the trust metric history data
|
||||
@@ -390,6 +234,8 @@ func (tm *TrustMetric) NextTimeInterval() {
|
||||
|
||||
// Copy returns a new trust metric with members containing the same values
|
||||
func (tm *TrustMetric) Copy() *TrustMetric {
|
||||
tm.mtx.Lock()
|
||||
defer tm.mtx.Unlock()
|
||||
if tm == nil {
|
||||
return nil
|
||||
}
|
||||
@@ -413,86 +259,22 @@ func (tm *TrustMetric) Copy() *TrustMetric {
|
||||
}
|
||||
}
|
||||
|
||||
// TrustMetricConfig - Configures the weight functions and time intervals for the metric
|
||||
type TrustMetricConfig struct {
|
||||
// Determines the percentage given to current behavior
|
||||
ProportionalWeight float64
|
||||
|
||||
// Determines the percentage given to prior behavior
|
||||
IntegralWeight float64
|
||||
|
||||
// The window of time that the trust metric will track events across.
|
||||
// This can be set to cover many days without issue
|
||||
TrackingWindow time.Duration
|
||||
|
||||
// Each interval should be short for adapability.
|
||||
// Less than 30 seconds is too sensitive,
|
||||
// and greater than 5 minutes will make the metric numb
|
||||
IntervalLength time.Duration
|
||||
}
|
||||
|
||||
// DefaultConfig returns a config with values that have been tested and produce desirable results
|
||||
func DefaultConfig() TrustMetricConfig {
|
||||
return TrustMetricConfig{
|
||||
ProportionalWeight: 0.4,
|
||||
IntegralWeight: 0.6,
|
||||
TrackingWindow: (time.Minute * 60 * 24) * 14, // 14 days.
|
||||
IntervalLength: 1 * time.Minute,
|
||||
}
|
||||
}
|
||||
|
||||
// NewMetric returns a trust metric with the default configuration
|
||||
func NewMetric() *TrustMetric {
|
||||
return NewMetricWithConfig(DefaultConfig())
|
||||
}
|
||||
|
||||
// NewMetricWithConfig returns a trust metric with a custom configuration
|
||||
func NewMetricWithConfig(tmc TrustMetricConfig) *TrustMetric {
|
||||
tm := new(TrustMetric)
|
||||
config := customConfig(tmc)
|
||||
|
||||
// Setup using the configuration values
|
||||
tm.proportionalWeight = config.ProportionalWeight
|
||||
tm.integralWeight = config.IntegralWeight
|
||||
tm.intervalLen = config.IntervalLength
|
||||
// The maximum number of time intervals is the tracking window / interval length
|
||||
tm.maxIntervals = int(config.TrackingWindow / tm.intervalLen)
|
||||
// The history size will be determined by the maximum number of time intervals
|
||||
tm.historyMaxSize = intervalToHistoryOffset(tm.maxIntervals) + 1
|
||||
// This metric has a perfect history so far
|
||||
tm.historyValue = 1.0
|
||||
// Setup the stop channel
|
||||
tm.stop = make(chan struct{})
|
||||
|
||||
go tm.processRequests()
|
||||
return tm
|
||||
}
|
||||
|
||||
/* Private methods */
|
||||
|
||||
// Ensures that all configuration elements have valid values
|
||||
func customConfig(tmc TrustMetricConfig) TrustMetricConfig {
|
||||
config := DefaultConfig()
|
||||
|
||||
// Check the config for set values, and setup appropriately
|
||||
if tmc.ProportionalWeight > 0 {
|
||||
config.ProportionalWeight = tmc.ProportionalWeight
|
||||
// This method is for a goroutine that handles all requests on the metric
|
||||
func (tm *TrustMetric) processRequests() {
|
||||
t := time.NewTicker(tm.intervalLen)
|
||||
defer t.Stop()
|
||||
loop:
|
||||
for {
|
||||
select {
|
||||
case <-t.C:
|
||||
tm.NextTimeInterval()
|
||||
case <-tm.stop:
|
||||
// Stop all further tracking for this metric
|
||||
break loop
|
||||
}
|
||||
}
|
||||
|
||||
if tmc.IntegralWeight > 0 {
|
||||
config.IntegralWeight = tmc.IntegralWeight
|
||||
}
|
||||
|
||||
if tmc.IntervalLength > time.Duration(0) {
|
||||
config.IntervalLength = tmc.IntervalLength
|
||||
}
|
||||
|
||||
if tmc.TrackingWindow > time.Duration(0) &&
|
||||
tmc.TrackingWindow >= config.IntervalLength {
|
||||
config.TrackingWindow = tmc.TrackingWindow
|
||||
}
|
||||
|
||||
return config
|
||||
}
|
||||
|
||||
// Wakes the trust metric up if it is currently paused
|
||||
@@ -508,9 +290,29 @@ func (tm *TrustMetric) unpause() {
|
||||
}
|
||||
}
|
||||
|
||||
// Calculates the derivative component
|
||||
func (tm *TrustMetric) derivativeValue() float64 {
|
||||
return tm.proportionalValue() - tm.historyValue
|
||||
// Calculates the trust value for the request processing
|
||||
func (tm *TrustMetric) calcTrustValue() float64 {
|
||||
weightedP := tm.proportionalWeight * tm.proportionalValue()
|
||||
weightedI := tm.integralWeight * tm.historyValue
|
||||
weightedD := tm.weightedDerivative()
|
||||
|
||||
tv := weightedP + weightedI + weightedD
|
||||
// Do not return a negative value.
|
||||
if tv < 0 {
|
||||
tv = 0
|
||||
}
|
||||
return tv
|
||||
}
|
||||
|
||||
// Calculates the current score for good/bad experiences
|
||||
func (tm *TrustMetric) proportionalValue() float64 {
|
||||
value := 1.0
|
||||
|
||||
total := tm.good + tm.bad
|
||||
if total > 0 {
|
||||
value = tm.good / total
|
||||
}
|
||||
return value
|
||||
}
|
||||
|
||||
// Strengthens the derivative component when the change is negative
|
||||
@@ -524,6 +326,35 @@ func (tm *TrustMetric) weightedDerivative() float64 {
|
||||
return weight * d
|
||||
}
|
||||
|
||||
// Calculates the derivative component
|
||||
func (tm *TrustMetric) derivativeValue() float64 {
|
||||
return tm.proportionalValue() - tm.historyValue
|
||||
}
|
||||
|
||||
// Calculates the integral (history) component of the trust value
|
||||
func (tm *TrustMetric) calcHistoryValue() float64 {
|
||||
var hv float64
|
||||
|
||||
for i := 0; i < tm.numIntervals; i++ {
|
||||
hv += tm.fadedMemoryValue(i) * tm.historyWeights[i]
|
||||
}
|
||||
|
||||
return hv / tm.historyWeightSum
|
||||
}
|
||||
|
||||
// Retrieves the actual history data value that represents the requested time interval
|
||||
func (tm *TrustMetric) fadedMemoryValue(interval int) float64 {
|
||||
first := tm.historySize - 1
|
||||
|
||||
if interval == 0 {
|
||||
// Base case
|
||||
return tm.history[first]
|
||||
}
|
||||
|
||||
offset := intervalToHistoryOffset(interval)
|
||||
return tm.history[first-offset]
|
||||
}
|
||||
|
||||
// Performs the update for our Faded Memories process, which allows the
|
||||
// trust metric tracking window to be large while maintaining a small
|
||||
// number of history data values
|
||||
@@ -550,68 +381,3 @@ func intervalToHistoryOffset(interval int) int {
|
||||
// the history data index = the floor of log2(i)
|
||||
return int(math.Floor(math.Log2(float64(interval))))
|
||||
}
|
||||
|
||||
// Retrieves the actual history data value that represents the requested time interval
|
||||
func (tm *TrustMetric) fadedMemoryValue(interval int) float64 {
|
||||
first := tm.historySize - 1
|
||||
|
||||
if interval == 0 {
|
||||
// Base case
|
||||
return tm.history[first]
|
||||
}
|
||||
|
||||
offset := intervalToHistoryOffset(interval)
|
||||
return tm.history[first-offset]
|
||||
}
|
||||
|
||||
// Calculates the integral (history) component of the trust value
|
||||
func (tm *TrustMetric) calcHistoryValue() float64 {
|
||||
var hv float64
|
||||
|
||||
for i := 0; i < tm.numIntervals; i++ {
|
||||
hv += tm.fadedMemoryValue(i) * tm.historyWeights[i]
|
||||
}
|
||||
|
||||
return hv / tm.historyWeightSum
|
||||
}
|
||||
|
||||
// Calculates the current score for good/bad experiences
|
||||
func (tm *TrustMetric) proportionalValue() float64 {
|
||||
value := 1.0
|
||||
|
||||
total := tm.good + tm.bad
|
||||
if total > 0 {
|
||||
value = tm.good / total
|
||||
}
|
||||
return value
|
||||
}
|
||||
|
||||
// Calculates the trust value for the request processing
|
||||
func (tm *TrustMetric) calcTrustValue() float64 {
|
||||
weightedP := tm.proportionalWeight * tm.proportionalValue()
|
||||
weightedI := tm.integralWeight * tm.historyValue
|
||||
weightedD := tm.weightedDerivative()
|
||||
|
||||
tv := weightedP + weightedI + weightedD
|
||||
// Do not return a negative value.
|
||||
if tv < 0 {
|
||||
tv = 0
|
||||
}
|
||||
return tv
|
||||
}
|
||||
|
||||
// This method is for a goroutine that handles all requests on the metric
|
||||
func (tm *TrustMetric) processRequests() {
|
||||
t := time.NewTicker(tm.intervalLen)
|
||||
defer t.Stop()
|
||||
loop:
|
||||
for {
|
||||
select {
|
||||
case <-t.C:
|
||||
tm.NextTimeInterval()
|
||||
case <-tm.stop:
|
||||
// Stop all further tracking for this metric
|
||||
break loop
|
||||
}
|
||||
}
|
||||
}
|
||||
90
p2p/trust/metric_test.go
Normal file
90
p2p/trust/metric_test.go
Normal file
@@ -0,0 +1,90 @@
|
||||
package trust
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
func TestTrustMetricScores(t *testing.T) {
|
||||
tm := NewMetric()
|
||||
|
||||
// Perfect score
|
||||
tm.GoodEvents(1)
|
||||
score := tm.TrustScore()
|
||||
assert.Equal(t, 100, score)
|
||||
|
||||
// Less than perfect score
|
||||
tm.BadEvents(10)
|
||||
score = tm.TrustScore()
|
||||
assert.NotEqual(t, 100, score)
|
||||
tm.Stop()
|
||||
}
|
||||
|
||||
func TestTrustMetricConfig(t *testing.T) {
|
||||
// 7 days
|
||||
window := time.Minute * 60 * 24 * 7
|
||||
config := TrustMetricConfig{
|
||||
TrackingWindow: window,
|
||||
IntervalLength: 2 * time.Minute,
|
||||
}
|
||||
|
||||
tm := NewMetricWithConfig(config)
|
||||
|
||||
// The max time intervals should be the TrackingWindow / IntervalLen
|
||||
assert.Equal(t, int(config.TrackingWindow/config.IntervalLength), tm.maxIntervals)
|
||||
|
||||
dc := DefaultConfig()
|
||||
// These weights should still be the default values
|
||||
assert.Equal(t, dc.ProportionalWeight, tm.proportionalWeight)
|
||||
assert.Equal(t, dc.IntegralWeight, tm.integralWeight)
|
||||
tm.Stop()
|
||||
|
||||
config.ProportionalWeight = 0.3
|
||||
config.IntegralWeight = 0.7
|
||||
tm = NewMetricWithConfig(config)
|
||||
|
||||
// These weights should be equal to our custom values
|
||||
assert.Equal(t, config.ProportionalWeight, tm.proportionalWeight)
|
||||
assert.Equal(t, config.IntegralWeight, tm.integralWeight)
|
||||
tm.Stop()
|
||||
}
|
||||
|
||||
func TestTrustMetricStopPause(t *testing.T) {
|
||||
// Cause time intervals to pass quickly
|
||||
config := TrustMetricConfig{
|
||||
TrackingWindow: 5 * time.Minute,
|
||||
IntervalLength: 10 * time.Millisecond,
|
||||
}
|
||||
|
||||
tm := NewMetricWithConfig(config)
|
||||
|
||||
// Allow some time intervals to pass and pause
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
tm.Pause()
|
||||
// Give the pause some time to take place
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
first := tm.Copy().numIntervals
|
||||
// Allow more time to pass and check the intervals are unchanged
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
assert.Equal(t, first, tm.numIntervals)
|
||||
|
||||
// Get the trust metric activated again
|
||||
tm.GoodEvents(5)
|
||||
// Allow some time intervals to pass and stop
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
tm.Stop()
|
||||
// Give the stop some time to take place
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
second := tm.Copy().numIntervals
|
||||
// Allow more time to pass and check the intervals are unchanged
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
assert.Equal(t, second, tm.numIntervals)
|
||||
|
||||
if first >= second {
|
||||
t.Fatalf("numIntervals should always increase or stay the same over time")
|
||||
}
|
||||
}
|
||||
192
p2p/trust/store.go
Normal file
192
p2p/trust/store.go
Normal file
@@ -0,0 +1,192 @@
|
||||
// Copyright 2017 Tendermint. All rights reserved.
|
||||
// Use of this source code is governed by Apache 2 LICENSE that can be found in the LICENSE file.
|
||||
|
||||
package trust
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
cmn "github.com/tendermint/tmlibs/common"
|
||||
dbm "github.com/tendermint/tmlibs/db"
|
||||
)
|
||||
|
||||
const defaultStorePeriodicSaveInterval = 1 * time.Minute
|
||||
|
||||
var trustMetricKey = []byte("trustMetricStore")
|
||||
|
||||
// TrustMetricStore - Manages all trust metrics for peers
|
||||
type TrustMetricStore struct {
|
||||
cmn.BaseService
|
||||
|
||||
// Maps a Peer.Key to that peer's TrustMetric
|
||||
peerMetrics map[string]*TrustMetric
|
||||
|
||||
// Mutex that protects the map and history data file
|
||||
mtx sync.Mutex
|
||||
|
||||
// The db where peer trust metric history data will be stored
|
||||
db dbm.DB
|
||||
|
||||
// This configuration will be used when creating new TrustMetrics
|
||||
config TrustMetricConfig
|
||||
}
|
||||
|
||||
// NewTrustMetricStore returns a store that saves data to the DB
|
||||
// and uses the config when creating new trust metrics
|
||||
func NewTrustMetricStore(db dbm.DB, tmc TrustMetricConfig) *TrustMetricStore {
|
||||
tms := &TrustMetricStore{
|
||||
peerMetrics: make(map[string]*TrustMetric),
|
||||
db: db,
|
||||
config: tmc,
|
||||
}
|
||||
|
||||
tms.BaseService = *cmn.NewBaseService(nil, "TrustMetricStore", tms)
|
||||
return tms
|
||||
}
|
||||
|
||||
// OnStart implements Service
|
||||
func (tms *TrustMetricStore) OnStart() error {
|
||||
if err := tms.BaseService.OnStart(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
tms.mtx.Lock()
|
||||
defer tms.mtx.Unlock()
|
||||
|
||||
tms.loadFromDB()
|
||||
go tms.saveRoutine()
|
||||
return nil
|
||||
}
|
||||
|
||||
// OnStop implements Service
|
||||
func (tms *TrustMetricStore) OnStop() {
|
||||
tms.BaseService.OnStop()
|
||||
|
||||
tms.mtx.Lock()
|
||||
defer tms.mtx.Unlock()
|
||||
|
||||
// Stop all trust metric go-routines
|
||||
for _, tm := range tms.peerMetrics {
|
||||
tm.Stop()
|
||||
}
|
||||
|
||||
// Make the final trust history data save
|
||||
tms.saveToDB()
|
||||
}
|
||||
|
||||
// Size returns the number of entries in the trust metric store
|
||||
func (tms *TrustMetricStore) Size() int {
|
||||
tms.mtx.Lock()
|
||||
defer tms.mtx.Unlock()
|
||||
|
||||
return tms.size()
|
||||
}
|
||||
|
||||
// GetPeerTrustMetric returns a trust metric by peer key
|
||||
func (tms *TrustMetricStore) GetPeerTrustMetric(key string) *TrustMetric {
|
||||
tms.mtx.Lock()
|
||||
defer tms.mtx.Unlock()
|
||||
|
||||
tm, ok := tms.peerMetrics[key]
|
||||
if !ok {
|
||||
// If the metric is not available, we will create it
|
||||
tm = NewMetricWithConfig(tms.config)
|
||||
// The metric needs to be in the map
|
||||
tms.peerMetrics[key] = tm
|
||||
}
|
||||
return tm
|
||||
}
|
||||
|
||||
// PeerDisconnected pauses the trust metric associated with the peer identified by the key
|
||||
func (tms *TrustMetricStore) PeerDisconnected(key string) {
|
||||
tms.mtx.Lock()
|
||||
defer tms.mtx.Unlock()
|
||||
|
||||
// If the Peer that disconnected has a metric, pause it
|
||||
if tm, ok := tms.peerMetrics[key]; ok {
|
||||
tm.Pause()
|
||||
}
|
||||
}
|
||||
|
||||
// Saves the history data for all peers to the store DB.
|
||||
// This public method acquires the trust metric store lock
|
||||
func (tms *TrustMetricStore) SaveToDB() {
|
||||
tms.mtx.Lock()
|
||||
defer tms.mtx.Unlock()
|
||||
|
||||
tms.saveToDB()
|
||||
}
|
||||
|
||||
/* Private methods */
|
||||
|
||||
// size returns the number of entries in the store without acquiring the mutex
|
||||
func (tms *TrustMetricStore) size() int {
|
||||
return len(tms.peerMetrics)
|
||||
}
|
||||
|
||||
/* Loading & Saving */
|
||||
/* Both loadFromDB and savetoDB assume the mutex has been acquired */
|
||||
|
||||
// Loads the history data for all peers from the store DB
|
||||
// cmn.Panics if file is corrupt
|
||||
func (tms *TrustMetricStore) loadFromDB() bool {
|
||||
// Obtain the history data we have so far
|
||||
bytes := tms.db.Get(trustMetricKey)
|
||||
if bytes == nil {
|
||||
return false
|
||||
}
|
||||
|
||||
peers := make(map[string]MetricHistoryJSON, 0)
|
||||
err := json.Unmarshal(bytes, &peers)
|
||||
if err != nil {
|
||||
cmn.PanicCrisis(cmn.Fmt("Could not unmarshal Trust Metric Store DB data: %v", err))
|
||||
}
|
||||
|
||||
// If history data exists in the file,
|
||||
// load it into trust metric
|
||||
for key, p := range peers {
|
||||
tm := NewMetricWithConfig(tms.config)
|
||||
|
||||
tm.Init(p)
|
||||
// Load the peer trust metric into the store
|
||||
tms.peerMetrics[key] = tm
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// Saves the history data for all peers to the store DB
|
||||
func (tms *TrustMetricStore) saveToDB() {
|
||||
tms.Logger.Debug("Saving TrustHistory to DB", "size", tms.size())
|
||||
|
||||
peers := make(map[string]MetricHistoryJSON, 0)
|
||||
|
||||
for key, tm := range tms.peerMetrics {
|
||||
// Add an entry for the peer identified by key
|
||||
peers[key] = tm.HistoryJSON()
|
||||
}
|
||||
|
||||
// Write all the data back to the DB
|
||||
bytes, err := json.Marshal(peers)
|
||||
if err != nil {
|
||||
tms.Logger.Error("Failed to encode the TrustHistory", "err", err)
|
||||
return
|
||||
}
|
||||
tms.db.SetSync(trustMetricKey, bytes)
|
||||
}
|
||||
|
||||
// Periodically saves the trust history data to the DB
|
||||
func (tms *TrustMetricStore) saveRoutine() {
|
||||
t := time.NewTicker(defaultStorePeriodicSaveInterval)
|
||||
defer t.Stop()
|
||||
loop:
|
||||
for {
|
||||
select {
|
||||
case <-t.C:
|
||||
tms.SaveToDB()
|
||||
case <-tms.Quit:
|
||||
break loop
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -150,85 +150,3 @@ func TestTrustMetricStorePeerScore(t *testing.T) {
|
||||
assert.NotEqual(t, 100, tm.TrustScore())
|
||||
tm.Stop()
|
||||
}
|
||||
|
||||
func TestTrustMetricScores(t *testing.T) {
|
||||
tm := NewMetric()
|
||||
|
||||
// Perfect score
|
||||
tm.GoodEvents(1)
|
||||
score := tm.TrustScore()
|
||||
assert.Equal(t, 100, score)
|
||||
|
||||
// Less than perfect score
|
||||
tm.BadEvents(10)
|
||||
score = tm.TrustScore()
|
||||
assert.NotEqual(t, 100, score)
|
||||
tm.Stop()
|
||||
}
|
||||
|
||||
func TestTrustMetricConfig(t *testing.T) {
|
||||
// 7 days
|
||||
window := time.Minute * 60 * 24 * 7
|
||||
config := TrustMetricConfig{
|
||||
TrackingWindow: window,
|
||||
IntervalLength: 2 * time.Minute,
|
||||
}
|
||||
|
||||
tm := NewMetricWithConfig(config)
|
||||
|
||||
// The max time intervals should be the TrackingWindow / IntervalLen
|
||||
assert.Equal(t, int(config.TrackingWindow/config.IntervalLength), tm.maxIntervals)
|
||||
|
||||
dc := DefaultConfig()
|
||||
// These weights should still be the default values
|
||||
assert.Equal(t, dc.ProportionalWeight, tm.proportionalWeight)
|
||||
assert.Equal(t, dc.IntegralWeight, tm.integralWeight)
|
||||
tm.Stop()
|
||||
|
||||
config.ProportionalWeight = 0.3
|
||||
config.IntegralWeight = 0.7
|
||||
tm = NewMetricWithConfig(config)
|
||||
|
||||
// These weights should be equal to our custom values
|
||||
assert.Equal(t, config.ProportionalWeight, tm.proportionalWeight)
|
||||
assert.Equal(t, config.IntegralWeight, tm.integralWeight)
|
||||
tm.Stop()
|
||||
}
|
||||
|
||||
func TestTrustMetricStopPause(t *testing.T) {
|
||||
// Cause time intervals to pass quickly
|
||||
config := TrustMetricConfig{
|
||||
TrackingWindow: 5 * time.Minute,
|
||||
IntervalLength: 10 * time.Millisecond,
|
||||
}
|
||||
|
||||
tm := NewMetricWithConfig(config)
|
||||
|
||||
// Allow some time intervals to pass and pause
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
tm.Pause()
|
||||
// Give the pause some time to take place
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
first := tm.Copy().numIntervals
|
||||
// Allow more time to pass and check the intervals are unchanged
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
assert.Equal(t, first, tm.numIntervals)
|
||||
|
||||
// Get the trust metric activated again
|
||||
tm.GoodEvents(5)
|
||||
// Allow some time intervals to pass and stop
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
tm.Stop()
|
||||
// Give the stop some time to take place
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
second := tm.Copy().numIntervals
|
||||
// Allow more time to pass and check the intervals are unchanged
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
assert.Equal(t, second, tm.numIntervals)
|
||||
|
||||
if first >= second {
|
||||
t.Fatalf("numIntervals should always increase or stay the same over time")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user