-
Notifications
You must be signed in to change notification settings - Fork 18
/
changeschecker.go
117 lines (93 loc) · 2.83 KB
/
changeschecker.go
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
package server
import (
"time"
"github.com/gelembjuk/oursql/lib/utils"
"github.com/gelembjuk/oursql/node/nodemanager"
)
type changesChecker struct {
S *NodeServer
logger *utils.LoggerMan
stopChan chan bool
completeChan chan bool
ticker int
lastCheckTime int64
}
func StartChangesChecker(s *NodeServer) (c *changesChecker) {
c = &changesChecker{}
c.logger = s.Logger
c.S = s
c.stopChan = make(chan bool) // to notify routine to stop
c.completeChan = make(chan bool) // routine to notify it stopped
c.ticker = 3
go c.Run()
return c
}
// Run function to request other nodes for changes regularly
func (c *changesChecker) Run() {
for {
c.logger.TraceExt.Printf("Check changes")
// check if it is time to exit or no
exit := false
select {
case <-c.stopChan:
exit = true
default:
}
if exit {
// exit
break
}
if c.ticker > 0 {
//c.logger.Trace.Printf("Changes Checker ticker value %d", c.ticker)
time.Sleep(1 * time.Second)
c.ticker = c.ticker - 1
continue
}
c.logger.TraceExt.Printf("Changes Checker. Go to check state")
pullResult, err := c.S.Node.GetCommunicationManager().CheckForChangesOnOtherNodes(c.lastCheckTime)
if err == nil {
c.processResults(pullResult)
}
// decide when to do next check
if c.S.Node.NodeNet.CheckHadInputConnects() {
// other nodes can connect to this node. No need to do extra check often
c.ticker = 180 // try again in 3 minutes
} else {
c.ticker = 5 // 5 seconds as it looks like other nodes can not connect to this node
}
//c.S.Node.NodeNet.StartNewSessionForInputConnects()
}
c.logger.Trace.Printf("Changes Checker Return routine")
c.completeChan <- true
}
func (c *changesChecker) Stop() error {
c.logger.Trace.Println("Stop changes checker")
close(c.stopChan) // notify routine to stop
c.logger.TraceExt.Println("Wait changes checker routine done")
// wait when it is stopped
<-c.completeChan
close(c.completeChan)
c.logger.TraceExt.Println("Changes Checker Stopped")
return nil
}
func (c changesChecker) processResults(res nodemanager.ChangesPullResults) {
if len(res.AddedTransactions) > 0 {
// if some transactions were added, notify to build new block
c.S.blocksMakerObj.DoNewBlock() // this will check state of the pool and start minting new block if there are enough
}
if res.AnyChangesPulled() {
c.logger.Trace.Println("Pull results")
c.logger.Trace.Printf("Transactions %d:", len(res.AddedTransactions))
for _, txID := range res.AddedTransactions {
c.logger.Trace.Printf(" %x", txID)
}
c.logger.Trace.Printf("Blocks %d:", len(res.AddedBlocks))
for _, bHash := range res.AddedBlocks {
c.logger.Trace.Printf(" %x", bHash)
}
c.logger.Trace.Printf("Nodes %d:", len(res.AddedNodes))
for _, n := range res.AddedNodes {
c.logger.Trace.Printf(" %s", n)
}
}
}