aboutsummaryrefslogtreecommitdiffstats
path: root/eth/sync.go
blob: dd7414da84bfe728b8b28ac5660afaa4b1d409b4 (plain) (blame)
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
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
package eth

import (
    "math"
    "sync/atomic"
    "time"

    "github.com/ethereum/go-ethereum/common"
    "github.com/ethereum/go-ethereum/core/types"
    "github.com/ethereum/go-ethereum/eth/downloader"
    "github.com/ethereum/go-ethereum/logger"
    "github.com/ethereum/go-ethereum/logger/glog"
)

const (
    forceSyncCycle      = 10 * time.Second       // Time interval to force syncs, even if few peers are available
    blockProcCycle      = 500 * time.Millisecond // Time interval to check for new blocks to process
    notifyCheckCycle    = 100 * time.Millisecond // Time interval to allow hash notifies to fulfill before hard fetching
    notifyArriveTimeout = 500 * time.Millisecond // Time allowance before an announced block is explicitly requested
    notifyFetchTimeout  = 5 * time.Second        // Maximum alloted time to return an explicitly requested block
    minDesiredPeerCount = 5                      // Amount of peers desired to start syncing
    blockProcAmount     = 256
)

// blockAnnounce is the hash notification of the availability of a new block in
// the network.
type blockAnnounce struct {
    hash common.Hash
    peer *peer
    time time.Time
}

// fetcher is responsible for collecting hash notifications, and periodically
// checking all unknown ones and individually fetching them.
func (pm *ProtocolManager) fetcher() {
    announces := make(map[common.Hash]*blockAnnounce)
    request := make(map[*peer][]common.Hash)
    pending := make(map[common.Hash]*blockAnnounce)
    cycle := time.Tick(notifyCheckCycle)

    // Iterate the block fetching until a quit is requested
    for {
        select {
        case notifications := <-pm.newHashCh:
            // A batch of hashes the notified, schedule them for retrieval
            glog.V(logger.Debug).Infof("Scheduling %d hash announcements from %s", len(notifications), notifications[0].peer.id)
            for _, announce := range notifications {
                announces[announce.hash] = announce
            }

        case <-cycle:
            // Clean up any expired block fetches
            for hash, announce := range pending {
                if time.Since(announce.time) > notifyFetchTimeout {
                    delete(pending, hash)
                }
            }
            // Check if any notified blocks failed to arrive
            for hash, announce := range announces {
                if time.Since(announce.time) > notifyArriveTimeout {
                    if !pm.chainman.HasBlock(hash) {
                        request[announce.peer] = append(request[announce.peer], hash)
                        pending[hash] = announce
                    }
                    delete(announces, hash)
                }
            }
            if len(request) == 0 {
                break
            }
            // Send out all block requests
            for peer, hashes := range request {
                glog.V(logger.Debug).Infof("Explicitly fetching %d blocks from %s", len(hashes), peer.id)
                peer.requestBlocks(hashes)
            }
            request = make(map[*peer][]common.Hash)

        case filter := <-pm.newBlockCh:
            // Blocks arrived, extract any explicit fetches, return all else
            var blocks types.Blocks
            select {
            case blocks = <-filter:
            case <-pm.quitSync:
                return
            }

            explicit, download := []*types.Block{}, []*types.Block{}
            for _, block := range blocks {
                hash := block.Hash()

                // Filter explicitly requested blocks from hash announcements
                if _, ok := pending[hash]; ok {
                    // Discard if already imported by other means
                    if !pm.chainman.HasBlock(hash) {
                        explicit = append(explicit, block)
                    } else {
                        delete(pending, hash)
                    }
                } else {
                    download = append(download, block)
                }
            }

            select {
            case filter <- download:
            case <-pm.quitSync:
                return
            }
            // If any explicit fetches were replied to, import them
            if count := len(explicit); count > 0 {
                glog.V(logger.Debug).Infof("Importing %d explicitly fetched blocks", count)
                go func() {
                    for _, block := range explicit {
                        hash := block.Hash()

                        // Make sure there's still something pending to import
                        if announce := pending[hash]; announce != nil {
                            delete(pending, hash)
                            if err := pm.importBlock(announce.peer, block, nil); err != nil {
                                glog.V(logger.Detail).Infof("Failed to import explicitly fetched block: %v", err)
                                return
                            }
                        }
                    }
                }()
            }

        case <-pm.quitSync:
            return
        }
    }
}

// syncer is responsible for periodically synchronising with the network, both
// downloading hashes and blocks as well as retrieving cached ones.
func (pm *ProtocolManager) syncer() {
    forceSync := time.Tick(forceSyncCycle)
    blockProc := time.Tick(blockProcCycle)
    blockProcPend := int32(0)

    for {
        select {
        case <-pm.newPeerCh:
            // Make sure we have peers to select from, then sync
            if pm.peers.Len() < minDesiredPeerCount {
                break
            }
            go pm.synchronise(pm.peers.BestPeer())

        case <-forceSync:
            // Force a sync even if not enough peers are present
            go pm.synchronise(pm.peers.BestPeer())

        case <-blockProc:
            // Try to pull some blocks from the downloaded
            if atomic.CompareAndSwapInt32(&blockProcPend, 0, 1) {
                go func() {
                    pm.processBlocks()
                    atomic.StoreInt32(&blockProcPend, 0)
                }()
            }

        case <-pm.quitSync:
            return
        }
    }
}

// processBlocks retrieves downloaded blocks from the download cache and tries
// to construct the local block chain with it. Note, since the block retrieval
// order matters, access to this function *must* be synchronized/serialized.
func (pm *ProtocolManager) processBlocks() error {
    pm.wg.Add(1)
    defer pm.wg.Done()

    // Short circuit if no blocks are available for insertion
    blocks := pm.downloader.TakeBlocks()
    if len(blocks) == 0 {
        return nil
    }
    glog.V(logger.Debug).Infof("Inserting chain with %d blocks (#%v - #%v)\n", len(blocks), blocks[0].RawBlock.Number(), blocks[len(blocks)-1].RawBlock.Number())

    for len(blocks) != 0 && !pm.quit {
        // Retrieve the first batch of blocks to insert
        max := int(math.Min(float64(len(blocks)), float64(blockProcAmount)))
        raw := make(types.Blocks, 0, max)
        for _, block := range blocks[:max] {
            raw = append(raw, block.RawBlock)
        }
        // Try to inset the blocks, drop the originating peer if there's an error
        index, err := pm.chainman.InsertChain(raw)
        if err != nil {
            glog.V(logger.Debug).Infoln("Downloaded block import failed:", err)
            pm.removePeer(blocks[index].OriginPeer)
            pm.downloader.Cancel()
            return err
        }
        blocks = blocks[max:]
    }
    return nil
}

// synchronise tries to sync up our local block chain with a remote peer, both
// adding various sanity checks as well as wrapping it with various log entries.
func (pm *ProtocolManager) synchronise(peer *peer) {
    // Short circuit if no peers are available
    if peer == nil {
        return
    }
    // Make sure the peer's TD is higher than our own. If not drop.
    if peer.td.Cmp(pm.chainman.Td()) <= 0 {
        return
    }
    // FIXME if we have the hash in our chain and the TD of the peer is
    // much higher than ours, something is wrong with us or the peer.
    // Check if the hash is on our own chain
    if pm.chainman.HasBlock(peer.recentHash) {
        glog.V(logger.Debug).Infoln("Synchronisation canceled: head already known")
        return
    }
    // Get the hashes from the peer (synchronously)
    glog.V(logger.Detail).Infof("Attempting synchronisation: %v, 0x%x", peer.id, peer.recentHash)

    err := pm.downloader.Synchronise(peer.id, peer.recentHash)
    switch err {
    case nil:
        glog.V(logger.Detail).Infof("Synchronisation completed")

    case downloader.ErrBusy:
        glog.V(logger.Detail).Infof("Synchronisation already in progress")

    case downloader.ErrTimeout, downloader.ErrBadPeer, downloader.ErrEmptyHashSet, downloader.ErrInvalidChain, downloader.ErrCrossCheckFailed:
        glog.V(logger.Debug).Infof("Removing peer %v: %v", peer.id, err)
        pm.removePeer(peer.id)

    case downloader.ErrPendingQueue:
        glog.V(logger.Debug).Infoln("Synchronisation aborted:", err)

    default:
        glog.V(logger.Warn).Infof("Synchronisation failed: %v", err)
    }
}