You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

207 lines
4.9 KiB

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