netsync/sync.go

   1 // Copyright 2015 The go-ethereum Authors
   2 // This file is part of the go-ethereum library.
   3 //
   4 // The go-ethereum library is free software: you can redistribute it and/or modify
   5 // it under the terms of the GNU Lesser General Public License as published by
   6 // the Free Software Foundation, either version 3 of the License, or
   7 // (at your option) any later version.
   8 //
   9 // The go-ethereum library is distributed in the hope that it will be useful,
  10 // but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  12 // GNU Lesser General Public License for more details.
  13 //
  14 // You should have received a copy of the GNU Lesser General Public License
  15 // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
  16
  17 package netsync
  18
  19 import (
  20         "math/rand"
  21         "sync/atomic"
  22         "time"
  23
  24         log "github.com/sirupsen/logrus"
  25
  26         "github.com/bytom/common"
  27         "github.com/bytom/protocol/bc/types"
  28 )
  29
  30 const (
  31         forceSyncCycle      = 10 * time.Second // Time interval to force syncs, even if few peers are available
  32         minDesiredPeerCount = 5                // Amount of peers desired to start syncing
  33
  34         // This is the target size for the packs of transactions sent by txsyncLoop.
  35         // A pack can get larger than this if a single transactions exceeds this size.
  36         txsyncPackSize = 100 * 1024
  37 )
  38
  39 type txsync struct {
  40         p   *peer
  41         txs []*types.Tx
  42 }
  43
  44 // syncer is responsible for periodically synchronising with the network, both
  45 // downloading hashes and blocks as well as handling the announcement handler.
  46 func (sm *SyncManager) syncer() {
  47         // Start and ensure cleanup of sync mechanisms
  48         sm.fetcher.Start()
  49         defer sm.fetcher.Stop()
  50         //defer sm.downloader.Terminate()
  51
  52         // Wait for different events to fire synchronisation operations
  53         forceSync := time.NewTicker(forceSyncCycle)
  54         defer forceSync.Stop()
  55
  56         for {
  57                 select {
  58                 case <-sm.newPeerCh:
  59                         log.Info("New peer connected.")
  60                         // Make sure we have peers to select from, then sync
  61                         if sm.sw.Peers().Size() < minDesiredPeerCount {
  62                                 break
  63                         }
  64                         go sm.synchronise()
  65
  66                 case <-forceSync.C:
  67                         // Force a sync even if not enough peers are present
  68                         go sm.synchronise()
  69
  70                 case <-sm.quitSync:
  71                         return
  72                 }
  73         }
  74 }
  75
  76 // synchronise tries to sync up our local block chain with a remote peer.
  77 func (sm *SyncManager) synchronise() {
  78         // Make sure only one goroutine is ever allowed past this point at once
  79         if !atomic.CompareAndSwapInt32(&sm.synchronising, 0, 1) {
  80                 log.Info("Synchronising ...")
  81                 return
  82         }
  83         defer atomic.StoreInt32(&sm.synchronising, 0)
  84
  85         peer, bestHeight := sm.peers.BestPeer()
  86         // Short circuit if no peers are available
  87         if peer == nil {
  88                 return
  89         }
  90         if bestHeight > sm.chain.Height() {
  91                 sm.blockKeeper.BlockRequestWorker(peer.Key, bestHeight)
  92         }
  93 }
  94
  95 // txsyncLoop takes care of the initial transaction sync for each new
  96 // connection. When a new peer appears, we relay all currently pending
  97 // transactions. In order to minimise egress bandwidth usage, we send
  98 // the transactions in small packs to one peer at a time.
  99 func (sm *SyncManager) txsyncLoop() {
 100         var (
 101                 pending = make(map[string]*txsync)
 102                 sending = false               // whether a send is active
 103                 pack    = new(txsync)         // the pack that is being sent
 104                 done    = make(chan error, 1) // result of the send
 105         )
 106
 107         // send starts a sending a pack of transactions from the sync.
 108         send := func(s *txsync) {
 109                 // Fill pack with transactions up to the target size.
 110                 size := common.StorageSize(0)
 111                 pack.p = s.p
 112                 pack.txs = pack.txs[:0]
 113                 for i := 0; i < len(s.txs) && size < txsyncPackSize; i++ {
 114                         pack.txs = append(pack.txs, s.txs[i])
 115                         size += common.StorageSize(s.txs[i].SerializedSize)
 116                 }
 117                 // Remove the transactions that will be sent.
 118                 s.txs = s.txs[:copy(s.txs, s.txs[len(pack.txs):])]
 119                 if len(s.txs) == 0 {
 120                         delete(pending, s.p.Key)
 121                 }
 122                 // Send the pack in the background.
 123                 log.Info("Sending batch of transactions. ", "count:", len(pack.txs), " bytes:", size)
 124                 sending = true
 125                 go func() { done <- pack.p.SendTransactions(pack.txs) }()
 126         }
 127
 128         // pick chooses the next pending sync.
 129         pick := func() *txsync {
 130                 if len(pending) == 0 {
 131                         return nil
 132                 }
 133                 n := rand.Intn(len(pending)) + 1
 134                 for _, s := range pending {
 135                         if n--; n == 0 {
 136                                 return s
 137                         }
 138                 }
 139                 return nil
 140         }
 141
 142         for {
 143                 select {
 144                 case s := <-sm.txSyncCh:
 145                         pending[s.p.Key] = s
 146                         if !sending {
 147                                 send(s)
 148                         }
 149                 case err := <-done:
 150                         sending = false
 151                         // Stop tracking peers that cause send failures.
 152                         if err != nil {
 153                                 log.Info("Transaction send failed", "err", err)
 154                                 delete(pending, pack.p.Key)
 155                         }
 156                         // Schedule the next send.
 157                         if s := pick(); s != nil {
 158                                 send(s)
 159                         }
 160                 case <-sm.quitSync:
 161                         return
 162                 }
 163         }
 164 }