netsync/protocol_reactor.go

   1 package netsync
   2
   3 import (
   4         "reflect"
   5         "strings"
   6         "sync"
   7         "time"
   8
   9         log "github.com/sirupsen/logrus"
  10         cmn "github.com/tendermint/tmlibs/common"
  11
  12         "github.com/bytom/errors"
  13         "github.com/bytom/p2p"
  14         "github.com/bytom/protocol"
  15         "github.com/bytom/protocol/bc"
  16         "github.com/bytom/protocol/bc/types"
  17 )
  18
  19 const (
  20         // BlockchainChannel is a channel for blocks and status updates
  21         BlockchainChannel        = byte(0x40)
  22         protocolHandshakeTimeout = time.Second * 10
  23         handshakeRetryTicker     = 4 * time.Second
  24 )
  25
  26 var (
  27         //ErrProtocolHandshakeTimeout peers handshake timeout
  28         ErrProtocolHandshakeTimeout = errors.New("Protocol handshake timeout")
  29         ErrStatusRequest            = errors.New("Status request error")
  30         ErrDiffGenesisHash          = errors.New("Different genesis hash")
  31 )
  32
  33 // Response describes the response standard.
  34 type Response struct {
  35         Status string      `json:"status,omitempty"`
  36         Msg    string      `json:"msg,omitempty"`
  37         Data   interface{} `json:"data,omitempty"`
  38 }
  39
  40 type initalPeerStatus struct {
  41         peerID      string
  42         height      uint64
  43         hash        *bc.Hash
  44         genesisHash *bc.Hash
  45 }
  46
  47 //ProtocolReactor handles new coming protocol message.
  48 type ProtocolReactor struct {
  49         p2p.BaseReactor
  50
  51         chain       *protocol.Chain
  52         blockKeeper *blockKeeper
  53         txPool      *protocol.TxPool
  54         sw          *p2p.Switch
  55         fetcher     *Fetcher
  56         peers       *peerSet
  57         handshakeMu sync.Mutex
  58         genesisHash bc.Hash
  59
  60         newPeerCh      chan struct{}
  61         quitReqBlockCh chan *string
  62         txSyncCh       chan *txsync
  63         peerStatusCh   chan *initalPeerStatus
  64 }
  65
  66 // NewProtocolReactor returns the reactor of whole blockchain.
  67 func NewProtocolReactor(chain *protocol.Chain, txPool *protocol.TxPool, sw *p2p.Switch, blockPeer *blockKeeper, fetcher *Fetcher, peers *peerSet, newPeerCh chan struct{}, txSyncCh chan *txsync, quitReqBlockCh chan *string) *ProtocolReactor {
  68         pr := &ProtocolReactor{
  69                 chain:          chain,
  70                 blockKeeper:    blockPeer,
  71                 txPool:         txPool,
  72                 sw:             sw,
  73                 fetcher:        fetcher,
  74                 peers:          peers,
  75                 newPeerCh:      newPeerCh,
  76                 txSyncCh:       txSyncCh,
  77                 quitReqBlockCh: quitReqBlockCh,
  78                 peerStatusCh:   make(chan *initalPeerStatus),
  79         }
  80         pr.BaseReactor = *p2p.NewBaseReactor("ProtocolReactor", pr)
  81         genesisBlock, _ := pr.chain.GetBlockByHeight(0)
  82         pr.genesisHash = genesisBlock.Hash()
  83
  84         return pr
  85 }
  86
  87 // GetChannels implements Reactor
  88 func (pr *ProtocolReactor) GetChannels() []*p2p.ChannelDescriptor {
  89         return []*p2p.ChannelDescriptor{
  90                 &p2p.ChannelDescriptor{
  91                         ID:                BlockchainChannel,
  92                         Priority:          5,
  93                         SendQueueCapacity: 100,
  94                 },
  95         }
  96 }
  97
  98 // OnStart implements BaseService
  99 func (pr *ProtocolReactor) OnStart() error {
 100         pr.BaseReactor.OnStart()
 101         return nil
 102 }
 103
 104 // OnStop implements BaseService
 105 func (pr *ProtocolReactor) OnStop() {
 106         pr.BaseReactor.OnStop()
 107 }
 108
 109 // syncTransactions starts sending all currently pending transactions to the given peer.
 110 func (pr *ProtocolReactor) syncTransactions(p *peer) {
 111         pending := pr.txPool.GetTransactions()
 112         if len(pending) == 0 {
 113                 return
 114         }
 115         txs := make([]*types.Tx, len(pending))
 116         for i, batch := range pending {
 117                 txs[i] = batch.Tx
 118         }
 119         pr.txSyncCh <- &txsync{p, txs}
 120 }
 121
 122 // AddPeer implements Reactor by sending our state to peer.
 123 func (pr *ProtocolReactor) AddPeer(peer *p2p.Peer) error {
 124         pr.handshakeMu.Lock()
 125         defer pr.handshakeMu.Unlock()
 126
 127         if ok := peer.Send(BlockchainChannel, struct{ BlockchainMessage }{&StatusRequestMessage{}}); !ok {
 128                 return ErrStatusRequest
 129         }
 130         retryTicker := time.Tick(handshakeRetryTicker)
 131         handshakeWait := time.NewTimer(protocolHandshakeTimeout)
 132         for {
 133                 select {
 134                 case status := <-pr.peerStatusCh:
 135                         if status.peerID == peer.Key {
 136                                 if strings.Compare(pr.genesisHash.String(), status.genesisHash.String()) != 0 {
 137                                         log.Info("Remote peer genesis block hash:", status.genesisHash.String(), " local hash:", pr.genesisHash.String())
 138                                         return ErrDiffGenesisHash
 139                                 }
 140                                 pr.peers.AddPeer(peer)
 141                                 pr.peers.SetPeerStatus(status.peerID, status.height, status.hash)
 142                                 pr.syncTransactions(pr.peers.Peer(peer.Key))
 143                                 pr.newPeerCh <- struct{}{}
 144                                 return nil
 145                         }
 146                 case <-retryTicker:
 147                         if ok := peer.Send(BlockchainChannel, struct{ BlockchainMessage }{&StatusRequestMessage{}}); !ok {
 148                                 return ErrStatusRequest
 149                         }
 150                 case <-handshakeWait.C:
 151                         return ErrProtocolHandshakeTimeout
 152                 }
 153         }
 154 }
 155
 156 // RemovePeer implements Reactor by removing peer from the pool.
 157 func (pr *ProtocolReactor) RemovePeer(peer *p2p.Peer, reason interface{}) {
 158         pr.quitReqBlockCh <- &peer.Key
 159         pr.peers.RemovePeer(peer.Key)
 160 }
 161
 162 // Receive implements Reactor by handling 4 types of messages (look below).
 163 func (pr *ProtocolReactor) Receive(chID byte, src *p2p.Peer, msgBytes []byte) {
 164         _, msg, err := DecodeMessage(msgBytes)
 165         if err != nil {
 166                 log.Errorf("Error decoding messagek %v", err)
 167                 return
 168         }
 169         log.WithFields(log.Fields{"peerID": src.Key, "msg": msg}).Info("Receive request")
 170
 171         switch msg := msg.(type) {
 172         case *BlockRequestMessage:
 173                 var block *types.Block
 174                 var err error
 175                 if msg.Height != 0 {
 176                         block, err = pr.chain.GetBlockByHeight(msg.Height)
 177                 } else {
 178                         block, err = pr.chain.GetBlockByHash(msg.GetHash())
 179                 }
 180                 if err != nil {
 181                         log.Errorf("Fail on BlockRequestMessage get block: %v", err)
 182                         return
 183                 }
 184                 response, err := NewBlockResponseMessage(block)
 185                 if err != nil {
 186                         log.Errorf("Fail on BlockRequestMessage create resoinse: %v", err)
 187                         return
 188                 }
 189                 src.TrySend(BlockchainChannel, struct{ BlockchainMessage }{response})
 190
 191         case *BlockResponseMessage:
 192                 log.Info("BlockResponseMessage height:", msg.GetBlock().Height)
 193                 pr.blockKeeper.AddBlock(msg.GetBlock(), src.Key)
 194
 195         case *StatusRequestMessage:
 196                 blockHeader := pr.chain.BestBlockHeader()
 197                 src.TrySend(BlockchainChannel, struct{ BlockchainMessage }{NewStatusResponseMessage(blockHeader, &pr.genesisHash)})
 198
 199         case *StatusResponseMessage:
 200                 peerStatus := &initalPeerStatus{
 201                         peerID:      src.Key,
 202                         height:      msg.Height,
 203                         hash:        msg.GetHash(),
 204                         genesisHash: msg.GetGenesisHash(),
 205                 }
 206                 pr.peerStatusCh <- peerStatus
 207
 208         case *TransactionNotifyMessage:
 209                 tx, err := msg.GetTransaction()
 210                 if err != nil {
 211                         log.Errorf("Error decoding new tx %v", err)
 212                         return
 213                 }
 214                 pr.blockKeeper.AddTx(tx, src.Key)
 215
 216         case *MineBlockMessage:
 217                 block, err := msg.GetMineBlock()
 218                 if err != nil {
 219                         log.Errorf("Error decoding mined block %v", err)
 220                         return
 221                 }
 222                 // Mark the peer as owning the block and schedule it for import
 223                 hash := block.Hash()
 224                 pr.peers.MarkBlock(src.Key, &hash)
 225                 pr.fetcher.Enqueue(src.Key, block)
 226                 pr.peers.SetPeerStatus(src.Key, block.Height, &hash)
 227
 228         default:
 229                 log.Error(cmn.Fmt("Unknown message type %v", reflect.TypeOf(msg)))
 230         }
 231 }