p2p/trust/store.go

   1 // Copyright 2017 Tendermint. All rights reserved.
   2 // Use of this source code is governed by Apache 2 LICENSE that can be found in the LICENSE file.
   3
   4 package trust
   5
   6 import (
   7         "encoding/json"
   8         "sync"
   9         "time"
  10
  11         cmn "github.com/tendermint/tmlibs/common"
  12         dbm "github.com/tendermint/tmlibs/db"
  13 )
  14
  15 const defaultStorePeriodicSaveInterval = 1 * time.Minute
  16
  17 var trustMetricKey = []byte("trustMetricStore")
  18
  19 // TrustMetricStore - Manages all trust metrics for peers
  20 type TrustMetricStore struct {
  21         cmn.BaseService
  22
  23         // Maps a Peer.Key to that peer's TrustMetric
  24         peerMetrics map[string]*TrustMetric
  25
  26         // Mutex that protects the map and history data file
  27         mtx sync.Mutex
  28
  29         // The db where peer trust metric history data will be stored
  30         db dbm.DB
  31
  32         // This configuration will be used when creating new TrustMetrics
  33         config TrustMetricConfig
  34 }
  35
  36 // NewTrustMetricStore returns a store that saves data to the DB
  37 // and uses the config when creating new trust metrics.
  38 // Use Start to to initialize the trust metric store
  39 func NewTrustMetricStore(db dbm.DB, tmc TrustMetricConfig) *TrustMetricStore {
  40         tms := &TrustMetricStore{
  41                 peerMetrics: make(map[string]*TrustMetric),
  42                 db:          db,
  43                 config:      tmc,
  44         }
  45
  46         tms.BaseService = *cmn.NewBaseService(nil, "TrustMetricStore", tms)
  47         return tms
  48 }
  49
  50 // OnStart implements Service
  51 func (tms *TrustMetricStore) OnStart() error {
  52         if err := tms.BaseService.OnStart(); err != nil {
  53                 return err
  54         }
  55
  56         tms.mtx.Lock()
  57         defer tms.mtx.Unlock()
  58
  59         tms.loadFromDB()
  60         go tms.saveRoutine()
  61         return nil
  62 }
  63
  64 // OnStop implements Service
  65 func (tms *TrustMetricStore) OnStop() {
  66         tms.BaseService.OnStop()
  67
  68         tms.mtx.Lock()
  69         defer tms.mtx.Unlock()
  70
  71         // Stop all trust metric go-routines
  72         for _, tm := range tms.peerMetrics {
  73                 tm.Stop()
  74         }
  75
  76         // Make the final trust history data save
  77         tms.saveToDB()
  78 }
  79
  80 // Size returns the number of entries in the trust metric store
  81 func (tms *TrustMetricStore) Size() int {
  82         tms.mtx.Lock()
  83         defer tms.mtx.Unlock()
  84
  85         return tms.size()
  86 }
  87
  88 // AddPeerTrustMetric takes an existing trust metric and associates it with a peer key.
  89 // The caller is expected to call Start on the TrustMetric being added
  90 func (tms *TrustMetricStore) AddPeerTrustMetric(key string, tm *TrustMetric) {
  91         tms.mtx.Lock()
  92         defer tms.mtx.Unlock()
  93
  94         if key == "" || tm == nil {
  95                 return
  96         }
  97         tms.peerMetrics[key] = tm
  98 }
  99
 100 // GetPeerTrustMetric returns a trust metric by peer key
 101 func (tms *TrustMetricStore) GetPeerTrustMetric(key string) *TrustMetric {
 102         tms.mtx.Lock()
 103         defer tms.mtx.Unlock()
 104
 105         tm, ok := tms.peerMetrics[key]
 106         if !ok {
 107                 // If the metric is not available, we will create it
 108                 tm = NewMetricWithConfig(tms.config)
 109                 tm.Start()
 110                 // The metric needs to be in the map
 111                 tms.peerMetrics[key] = tm
 112         }
 113         return tm
 114 }
 115
 116 // PeerDisconnected pauses the trust metric associated with the peer identified by the key
 117 func (tms *TrustMetricStore) PeerDisconnected(key string) {
 118         tms.mtx.Lock()
 119         defer tms.mtx.Unlock()
 120
 121         // If the Peer that disconnected has a metric, pause it
 122         if tm, ok := tms.peerMetrics[key]; ok {
 123                 tm.Pause()
 124         }
 125 }
 126
 127 // Saves the history data for all peers to the store DB.
 128 // This public method acquires the trust metric store lock
 129 func (tms *TrustMetricStore) SaveToDB() {
 130         tms.mtx.Lock()
 131         defer tms.mtx.Unlock()
 132
 133         tms.saveToDB()
 134 }
 135
 136 /* Private methods */
 137
 138 // size returns the number of entries in the store without acquiring the mutex
 139 func (tms *TrustMetricStore) size() int {
 140         return len(tms.peerMetrics)
 141 }
 142
 143 /* Loading & Saving */
 144 /* Both loadFromDB and savetoDB assume the mutex has been acquired */
 145
 146 // Loads the history data for all peers from the store DB
 147 // cmn.Panics if file is corrupt
 148 func (tms *TrustMetricStore) loadFromDB() bool {
 149         // Obtain the history data we have so far
 150         bytes := tms.db.Get(trustMetricKey)
 151         if bytes == nil {
 152                 return false
 153         }
 154
 155         peers := make(map[string]MetricHistoryJSON)
 156         err := json.Unmarshal(bytes, &peers)
 157         if err != nil {
 158                 cmn.PanicCrisis(cmn.Fmt("Could not unmarshal Trust Metric Store DB data: %v", err))
 159         }
 160
 161         // If history data exists in the file,
 162         // load it into trust metric
 163         for key, p := range peers {
 164                 tm := NewMetricWithConfig(tms.config)
 165
 166                 tm.Start()
 167                 tm.Init(p)
 168                 // Load the peer trust metric into the store
 169                 tms.peerMetrics[key] = tm
 170         }
 171         return true
 172 }
 173
 174 // Saves the history data for all peers to the store DB
 175 func (tms *TrustMetricStore) saveToDB() {
 176         tms.Logger.Debug("Saving TrustHistory to DB", "size", tms.size())
 177
 178         peers := make(map[string]MetricHistoryJSON)
 179
 180         for key, tm := range tms.peerMetrics {
 181                 // Add an entry for the peer identified by key
 182                 peers[key] = tm.HistoryJSON()
 183         }
 184
 185         // Write all the data back to the DB
 186         bytes, err := json.Marshal(peers)
 187         if err != nil {
 188                 tms.Logger.Error("Failed to encode the TrustHistory", "err", err)
 189                 return
 190         }
 191         tms.db.SetSync(trustMetricKey, bytes)
 192 }
 193
 194 // Periodically saves the trust history data to the DB
 195 func (tms *TrustMetricStore) saveRoutine() {
 196         t := time.NewTicker(defaultStorePeriodicSaveInterval)
 197         defer t.Stop()
 198 loop:
 199         for {
 200                 select {
 201                 case <-t.C:
 202                         tms.SaveToDB()
 203                 case <-tms.Quit:
 204                         break loop
 205                 }
 206         }
 207 }