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         log.Info("bk peer num:", sm.blockKeeper.peers.Len(), " sw peer num:", sm.sw.Peers().Size(), " ", sm.sw.Peers().List())
  79         // Make sure only one goroutine is ever allowed past this point at once
  80         if !atomic.CompareAndSwapInt32(&sm.synchronising, 0, 1) {
  81                 log.Info("Synchronising ...")
  82                 return
  83         }
  84         defer atomic.StoreInt32(&sm.synchronising, 0)
  85
  86         peer, bestHeight := sm.peers.BestPeer()
  87         // Short circuit if no peers are available
  88         if peer == nil {
  89                 return
  90         }
  91
  92         if ok := sm.Switch().Peers().Has(peer.Key); !ok {
  93                 log.Info("Peer disconnected")
  94                 sm.sw.StopPeerGracefully(peer)
  95                 return
  96         }
  97
  98         if bestHeight > sm.chain.BestBlockHeight() {
  99                 log.Info("sync peer:", peer.Addr(), " height:", bestHeight)
 100                 sm.blockKeeper.BlockRequestWorker(peer.Key, bestHeight)
 101         }
 102 }
 103
 104 // txsyncLoop takes care of the initial transaction sync for each new
 105 // connection. When a new peer appears, we relay all currently pending
 106 // transactions. In order to minimise egress bandwidth usage, we send
 107 // the transactions in small packs to one peer at a time.
 108 func (sm *SyncManager) txsyncLoop() {
 109         var (
 110                 pending = make(map[string]*txsync)
 111                 sending = false               // whether a send is active
 112                 pack    = new(txsync)         // the pack that is being sent
 113                 done    = make(chan error, 1) // result of the send
 114         )
 115
 116         // send starts a sending a pack of transactions from the sync.
 117         send := func(s *txsync) {
 118                 // Fill pack with transactions up to the target size.
 119                 size := common.StorageSize(0)
 120                 pack.p = s.p
 121                 pack.txs = pack.txs[:0]
 122                 for i := 0; i < len(s.txs) && size < txsyncPackSize; i++ {
 123                         pack.txs = append(pack.txs, s.txs[i])
 124                         size += common.StorageSize(s.txs[i].SerializedSize)
 125                 }
 126                 // Remove the transactions that will be sent.
 127                 s.txs = s.txs[:copy(s.txs, s.txs[len(pack.txs):])]
 128                 if len(s.txs) == 0 {
 129                         delete(pending, s.p.Key)
 130                 }
 131                 // Send the pack in the background.
 132                 log.Info("Sending batch of transactions. ", "count:", len(pack.txs), " bytes:", size)
 133                 sending = true
 134                 go func() { done <- pack.p.SendTransactions(pack.txs) }()
 135         }
 136
 137         // pick chooses the next pending sync.
 138         pick := func() *txsync {
 139                 if len(pending) == 0 {
 140                         return nil
 141                 }
 142                 n := rand.Intn(len(pending)) + 1
 143                 for _, s := range pending {
 144                         if n--; n == 0 {
 145                                 return s
 146                         }
 147                 }
 148                 return nil
 149         }
 150
 151         for {
 152                 select {
 153                 case s := <-sm.txSyncCh:
 154                         pending[s.p.Key] = s
 155                         if !sending {
 156                                 send(s)
 157                         }
 158                 case err := <-done:
 159                         sending = false
 160                         // Stop tracking peers that cause send failures.
 161                         if err != nil {
 162                                 log.Info("Transaction send failed", "err", err)
 163                                 delete(pending, pack.p.Key)
 164                         }
 165                         // Schedule the next send.
 166                         if s := pick(); s != nil {
 167                                 send(s)
 168                         }
 169                 case <-sm.quitSync:
 170                         return
 171                 }
 172         }
 173 }