From 7171823fc6d667904cfdc926a51f2848a804a04d Mon Sep 17 00:00:00 2001 From: Jae Kwon Date: Wed, 25 Mar 2015 17:16:49 -0700 Subject: [PATCH] Fix blockpool bugs and clean up log messages. --- blockchain/pool.go | 21 ++++++--------- blockchain/reactor.go | 61 +++++++++++++++++++++++-------------------- consensus/reactor.go | 2 +- p2p/addrbook.go | 2 +- p2p/connection.go | 1 + p2p/pex_reactor.go | 10 ++++--- p2p/switch.go | 2 +- 7 files changed, 51 insertions(+), 48 deletions(-) diff --git a/blockchain/pool.go b/blockchain/pool.go index 577701519..0bd620463 100644 --- a/blockchain/pool.go +++ b/blockchain/pool.go @@ -10,14 +10,12 @@ import ( ) const ( - maxOutstandingRequestsPerPeer = 10 - inputsChannelCapacity = 100 - maxTries = 3 - requestIntervalMS = 500 - requestBatchSize = 50 - maxPendingRequests = 50 - maxTotalRequests = 100 - maxRequestsPerPeer = 20 + maxTries = 3 + inputsChannelCapacity = 200 + requestIntervalMS = 500 + maxPendingRequests = 200 + maxTotalRequests = 300 + maxRequestsPerPeer = 300 ) var ( @@ -85,9 +83,7 @@ RUN_LOOP: if atomic.LoadInt32(&pool.running) == 0 { break RUN_LOOP } - height, numPending, numTotal := pool.GetStatus() - log.Debug("BlockPool.run", "height", height, "numPending", numPending, - "numTotal", numTotal) + _, numPending, numTotal := pool.GetStatus() if numPending >= maxPendingRequests { // sleep for a bit. time.Sleep(requestIntervalMS * time.Millisecond) @@ -344,14 +340,13 @@ func requestRoutine(pool *BlockPool, height uint) { } peer = pool.pickIncrAvailablePeer(height) if peer == nil { - log.Debug("No peers available", "height", height) + //log.Debug("No peers available", "height", height) time.Sleep(requestIntervalMS * time.Millisecond) continue PICK_LOOP } break PICK_LOOP } - log.Debug("Selected peer for request", "height", height, "peerId", peer.id) pool.setPeerForRequest(height, peer.id) for try := 0; try < maxTries; try++ { diff --git a/blockchain/reactor.go b/blockchain/reactor.go index a3e93e989..b0ae92455 100644 --- a/blockchain/reactor.go +++ b/blockchain/reactor.go @@ -90,16 +90,15 @@ func (bcR *BlockchainReactor) GetChannels() []*p2p.ChannelDescriptor { &p2p.ChannelDescriptor{ Id: BlockchainChannel, Priority: 5, - SendQueueCapacity: 20, // Queue 20 blocks to send to a peer. + SendQueueCapacity: 100, }, } } // Implements Reactor func (bcR *BlockchainReactor) AddPeer(peer *p2p.Peer) { - log.Debug("BlockchainReactor AddPeer", "peer", peer) // Send peer our state. - peer.Send(BlockchainChannel, PeerStatusMessage{bcR.store.Height()}) + peer.Send(BlockchainChannel, bcPeerStatusMessage{bcR.store.Height()}) } // Implements Reactor @@ -115,14 +114,15 @@ func (bcR *BlockchainReactor) Receive(chId byte, src *p2p.Peer, msgBytes []byte) log.Warn("Error decoding message", "error", err) return } - log.Debug("BlockchainReactor received message", "msg", msg_) + + log.Info("Received message", "msg", msg_) switch msg := msg_.(type) { - case BlockRequestMessage: + case bcBlockRequestMessage: // Got a request for a block. Respond with block if we have it. block := bcR.store.LoadBlock(msg.Height) if block != nil { - msg := BlockResponseMessage{Block: block} + msg := bcBlockResponseMessage{Block: block} queued := src.TrySend(BlockchainChannel, msg) if !queued { // queue is full, just ignore. @@ -130,10 +130,10 @@ func (bcR *BlockchainReactor) Receive(chId byte, src *p2p.Peer, msgBytes []byte) } else { // TODO peer is asking for things we don't have. } - case BlockResponseMessage: + case bcBlockResponseMessage: // Got a block. bcR.pool.AddBlock(msg.Block, src.Key) - case PeerStatusMessage: + case bcPeerStatusMessage: // Got a peer status. bcR.pool.SetPeerHeight(src.Key, msg.Height) default: @@ -155,7 +155,7 @@ FOR_LOOP: // We can't fulfill the request. continue FOR_LOOP } - msg := BlockRequestMessage{request.Height} + msg := bcBlockRequestMessage{request.Height} queued := peer.TrySend(BlockchainChannel, msg) if !queued { // We couldn't queue the request. @@ -174,24 +174,29 @@ FOR_LOOP: for i := 0; i < 10; i++ { // See if there are any blocks to sync. first, second := bcR.pool.PeekTwoBlocks() + //log.Debug("TrySync peeked", "first", first, "second", second) if first == nil || second == nil { // We need both to sync the first block. break SYNC_LOOP } - firstParts := first.MakePartSet().Header() + firstParts := first.MakePartSet() + firstPartsHeader := firstParts.Header() // Finally, verify the first block using the second's validation. err := bcR.state.BondedValidators.VerifyValidation( - first.Hash(), firstParts, first.Height, second.Validation) + first.Hash(), firstPartsHeader, first.Height, second.Validation) if err != nil { + log.Debug("error in validation", "error", err) bcR.pool.RedoRequest(first.Height) break SYNC_LOOP } else { bcR.pool.PopRequest() - err := bcR.state.AppendBlock(first, firstParts) + err := bcR.state.AppendBlock(first, firstPartsHeader) if err != nil { // TODO This is bad, are we zombie? panic(Fmt("Failed to process committed block: %v", err)) } + bcR.store.SaveBlock(first, firstParts, second.Validation) + bcR.state.Save() lastValidatedBlock = first } } @@ -224,7 +229,7 @@ FOR_LOOP: } func (bcR *BlockchainReactor) BroadcastStatus() error { - bcR.sw.Broadcast(BlockchainChannel, PeerStatusMessage{bcR.store.Height()}) + bcR.sw.Broadcast(BlockchainChannel, bcPeerStatusMessage{bcR.store.Height()}) return nil } @@ -245,11 +250,11 @@ func DecodeMessage(bz []byte) (msgType byte, msg interface{}, err error) { r := bytes.NewReader(bz) switch msgType { case msgTypeBlockRequest: - msg = binary.ReadBinary(BlockRequestMessage{}, r, n, &err) + msg = binary.ReadBinary(bcBlockRequestMessage{}, r, n, &err) case msgTypeBlockResponse: - msg = binary.ReadBinary(BlockResponseMessage{}, r, n, &err) + msg = binary.ReadBinary(bcBlockResponseMessage{}, r, n, &err) case msgTypePeerStatus: - msg = binary.ReadBinary(PeerStatusMessage{}, r, n, &err) + msg = binary.ReadBinary(bcPeerStatusMessage{}, r, n, &err) default: msg = nil } @@ -258,36 +263,36 @@ func DecodeMessage(bz []byte) (msgType byte, msg interface{}, err error) { //------------------------------------- -type BlockRequestMessage struct { +type bcBlockRequestMessage struct { Height uint } -func (m BlockRequestMessage) TypeByte() byte { return msgTypeBlockRequest } +func (m bcBlockRequestMessage) TypeByte() byte { return msgTypeBlockRequest } -func (m BlockRequestMessage) String() string { - return fmt.Sprintf("[BlockRequestMessage %v]", m.Height) +func (m bcBlockRequestMessage) String() string { + return fmt.Sprintf("[bcBlockRequestMessage %v]", m.Height) } //------------------------------------- -type BlockResponseMessage struct { +type bcBlockResponseMessage struct { Block *types.Block } -func (m BlockResponseMessage) TypeByte() byte { return msgTypeBlockResponse } +func (m bcBlockResponseMessage) TypeByte() byte { return msgTypeBlockResponse } -func (m BlockResponseMessage) String() string { - return fmt.Sprintf("[BlockResponseMessage %v]", m.Block.Height) +func (m bcBlockResponseMessage) String() string { + return fmt.Sprintf("[bcBlockResponseMessage %v]", m.Block.Height) } //------------------------------------- -type PeerStatusMessage struct { +type bcPeerStatusMessage struct { Height uint } -func (m PeerStatusMessage) TypeByte() byte { return msgTypePeerStatus } +func (m bcPeerStatusMessage) TypeByte() byte { return msgTypePeerStatus } -func (m PeerStatusMessage) String() string { - return fmt.Sprintf("[PeerStatusMessage %v]", m.Height) +func (m bcPeerStatusMessage) String() string { + return fmt.Sprintf("[bcPeerStatusMessage %v]", m.Height) } diff --git a/consensus/reactor.go b/consensus/reactor.go index 8d992bba7..a3028b6d5 100644 --- a/consensus/reactor.go +++ b/consensus/reactor.go @@ -69,7 +69,7 @@ func (conR *ConsensusReactor) Stop() { } func (conR *ConsensusReactor) IsRunning() bool { - return atomic.LoadUint32(&conR.running) == 0 + return atomic.LoadUint32(&conR.running) == 1 } // Implements Reactor diff --git a/p2p/addrbook.go b/p2p/addrbook.go index 43cad607c..893cf1c9e 100644 --- a/p2p/addrbook.go +++ b/p2p/addrbook.go @@ -381,7 +381,7 @@ out: for { select { case <-dumpAddressTicker.C: - log.Debug("Saving book to file", "size", a.Size()) + log.Debug("Saving AddrBook to file", "size", a.Size()) a.saveToFile(a.filePath) case <-a.quit: break out diff --git a/p2p/connection.go b/p2p/connection.go index 0e26480f1..c75538365 100644 --- a/p2p/connection.go +++ b/p2p/connection.go @@ -417,6 +417,7 @@ FOR_LOOP: } msgBytes := channel.recvMsgPacket(pkt) if msgBytes != nil { + log.Debug("Received bytes", "chId", pkt.ChannelId, "msgBytes", msgBytes) c.onReceive(pkt.ChannelId, msgBytes) } default: diff --git a/p2p/pex_reactor.go b/p2p/pex_reactor.go index 439b4cf9b..5ba238d31 100644 --- a/p2p/pex_reactor.go +++ b/p2p/pex_reactor.go @@ -176,10 +176,12 @@ func (pexR *PEXReactor) ensurePeers() { alreadyDialing := pexR.sw.IsDialing(try) alreadyConnected := pexR.sw.Peers().Has(try.String()) if alreadySelected || alreadyDialing || alreadyConnected { - log.Debug("Cannot dial address", "addr", try, - "alreadySelected", alreadySelected, - "alreadyDialing", alreadyDialing, - "alreadyConnected", alreadyConnected) + /* + log.Debug("Cannot dial address", "addr", try, + "alreadySelected", alreadySelected, + "alreadyDialing", alreadyDialing, + "alreadyConnected", alreadyConnected) + */ continue } else { log.Debug("Will dial address", "addr", try) diff --git a/p2p/switch.go b/p2p/switch.go index 105561314..1eb513089 100644 --- a/p2p/switch.go +++ b/p2p/switch.go @@ -132,7 +132,7 @@ func (sw *Switch) AddPeerWithConnection(conn net.Conn, outbound bool) (*Peer, er } // Start the peer - go peer.start() + peer.start() // Notify listeners. sw.doAddPeer(peer)