From ba43364f36db690528cc62196969414ef5e98833 Mon Sep 17 00:00:00 2001 From: obscuren Date: Wed, 24 Sep 2014 11:41:57 +0200 Subject: tmp --- peer.go | 97 +++++++++++++++++++++++++++++++++++++++-------------------------- 1 file changed, 59 insertions(+), 38 deletions(-) (limited to 'peer.go') diff --git a/peer.go b/peer.go index 529a35492..8224f4831 100644 --- a/peer.go +++ b/peer.go @@ -131,6 +131,7 @@ type Peer struct { // Last received pong message lastPong int64 lastBlockReceived time.Time + LastHashReceived time.Time host []byte port uint16 @@ -176,6 +177,7 @@ func NewPeer(conn net.Conn, ethereum *Ethereum, inbound bool) *Peer { caps: ethereum.ServerCaps(), version: ethereum.ClientIdentity().String(), protocolCaps: ethutil.NewValue(nil), + td: big.NewInt(0), } } @@ -191,6 +193,7 @@ func NewOutboundPeer(addr string, ethereum *Ethereum, caps Caps) *Peer { caps: caps, version: ethereum.ClientIdentity().String(), protocolCaps: ethutil.NewValue(nil), + td: big.NewInt(0), } // Set up the connection in another goroutine so we don't block the main thread @@ -505,6 +508,9 @@ func (p *Peer) HandleInbound() { for it.Next() { hash := it.Value().Bytes() + p.lastReceivedHash = hash + p.LastHashReceived = time.Now() + if blockPool.HasCommonHash(hash) { foundCommonHash = true @@ -512,15 +518,16 @@ func (p *Peer) HandleInbound() { } blockPool.AddHash(hash) - - p.lastReceivedHash = hash - - p.lastBlockReceived = time.Now() } - if foundCommonHash || msg.Data.Len() == 0 { - p.FetchBlocks() - } else { + /* + if foundCommonHash || msg.Data.Len() == 0 { + p.FetchBlocks() + } else { + p.FetchHashes() + } + */ + if !foundCommonHash && msg.Data.Len() != 0 { p.FetchHashes() } @@ -539,19 +546,21 @@ func (p *Peer) HandleInbound() { p.lastBlockReceived = time.Now() } - var err error - blockPool.CheckLinkAndProcess(func(block *ethchain.Block) { - err = p.ethereum.StateManager().Process(block, false) - }) - - if err != nil { - peerlogger.Infoln(err) - } else { - // Don't trigger if there's just one block. - if blockPool.Len() != 0 && msg.Data.Len() > 1 { - p.FetchBlocks() + /* + var err error + blockPool.CheckLinkAndProcess(func(block *ethchain.Block) { + err = p.ethereum.StateManager().Process(block, false) + }) + + if err != nil { + peerlogger.Infoln(err) + } else { + // Don't trigger if there's just one block. + if blockPool.Len() != 0 && msg.Data.Len() > 1 { + p.FetchBlocks() + } } - } + */ } } } @@ -560,10 +569,7 @@ func (p *Peer) HandleInbound() { p.Stop() } -func (self *Peer) FetchBlocks() { - blockPool := self.ethereum.blockPool - - hashes := blockPool.Take(100, self) +func (self *Peer) FetchBlocks(hashes [][]byte) { if len(hashes) > 0 { self.QueueMessage(ethwire.NewMessage(ethwire.MsgGetBlocksTy, ethutil.ByteSliceToInterface(hashes))) } @@ -572,7 +578,7 @@ func (self *Peer) FetchBlocks() { func (self *Peer) FetchHashes() { blockPool := self.ethereum.blockPool - if self.td.Cmp(blockPool.td) >= 0 { + if self.td.Cmp(self.ethereum.HighestTDPeer()) >= 0 { blockPool.td = self.td if !blockPool.HasLatestHash() { @@ -581,6 +587,10 @@ func (self *Peer) FetchHashes() { } } +func (self *Peer) FetchingHashes() bool { + return time.Since(self.LastHashReceived) < 5*time.Second +} + // General update method func (self *Peer) update() { serviceTimer := time.NewTicker(5 * time.Second) @@ -589,11 +599,22 @@ out: for { select { case <-serviceTimer.C: - since := time.Since(self.lastBlockReceived) - if since > 10*time.Second && self.ethereum.blockPool.Len() != 0 && self.IsCap("eth") { - self.FetchHashes() - } else if since > 5*time.Second { - self.catchingUp = false + if self.IsCap("eth") { + var ( + sinceBlock = time.Since(self.lastBlockReceived) + sinceHash = time.Since(self.LastHashReceived) + ) + + if sinceBlock > 5*time.Second && sinceHash > 5*time.Second { + self.catchingUp = false + } + + if sinceHash > 10*time.Second && self.ethereum.blockPool.Len() != 0 { + // XXX While this is completely and utterly incorrect, in order to do anything on the test net is to do it this way + // Assume that when fetching hashes timeouts, we are done. + //self.FetchHashes() + //self.FetchBlocks() + } } case <-self.quit: break out @@ -761,6 +782,14 @@ func (p *Peer) handleHandshake(msg *ethwire.Msg) { return } + // Self connect detection + pubkey := p.ethereum.KeyManager().PublicKey() + if bytes.Compare(pubkey[1:], pub) == 0 { + p.Stop() + + return + } + usedPub := 0 // This peer is already added to the peerlist so we expect to find a double pubkey at least once eachPeer(p.ethereum.Peers(), func(peer *Peer, e *list.Element) { @@ -779,16 +808,8 @@ func (p *Peer) handleHandshake(msg *ethwire.Msg) { // If this is an inbound connection send an ack back if p.inbound { p.port = uint16(port) - - // Self connect detection - pubkey := p.ethereum.KeyManager().PublicKey() - if bytes.Compare(pubkey, p.pubkey) == 0 { - p.Stop() - - return - } - } + p.SetVersion(clientId) p.versionKnown = true -- cgit From b66fcf85dfecb13c3ed2b9f46ad6bb257ce84411 Mon Sep 17 00:00:00 2001 From: obscuren Date: Wed, 24 Sep 2014 11:55:02 +0200 Subject: checkp --- peer.go | 53 ++++++++++++++++++++++++++++++++--------------------- 1 file changed, 32 insertions(+), 21 deletions(-) (limited to 'peer.go') diff --git a/peer.go b/peer.go index 529a35492..00987c123 100644 --- a/peer.go +++ b/peer.go @@ -131,6 +131,7 @@ type Peer struct { // Last received pong message lastPong int64 lastBlockReceived time.Time + lastHashReceived time.Time host []byte port uint16 @@ -505,6 +506,9 @@ func (p *Peer) HandleInbound() { for it.Next() { hash := it.Value().Bytes() + p.lastReceivedHash = hash + p.lastHashReceived = time.Now() + if blockPool.HasCommonHash(hash) { foundCommonHash = true @@ -512,10 +516,6 @@ func (p *Peer) HandleInbound() { } blockPool.AddHash(hash) - - p.lastReceivedHash = hash - - p.lastBlockReceived = time.Now() } if foundCommonHash || msg.Data.Len() == 0 { @@ -546,12 +546,12 @@ func (p *Peer) HandleInbound() { if err != nil { peerlogger.Infoln(err) - } else { + } /*else { // Don't trigger if there's just one block. if blockPool.Len() != 0 && msg.Data.Len() > 1 { p.FetchBlocks() } - } + }*/ } } } @@ -583,17 +583,28 @@ func (self *Peer) FetchHashes() { // General update method func (self *Peer) update() { - serviceTimer := time.NewTicker(5 * time.Second) + serviceTimer := time.NewTicker(100 * time.Millisecond) out: for { select { case <-serviceTimer.C: - since := time.Since(self.lastBlockReceived) - if since > 10*time.Second && self.ethereum.blockPool.Len() != 0 && self.IsCap("eth") { - self.FetchHashes() - } else if since > 5*time.Second { - self.catchingUp = false + if self.IsCap("eth") { + var ( + sinceBlock = time.Since(self.lastBlockReceived) + sinceHash = time.Since(self.lastHashReceived) + ) + + if sinceBlock > 5*time.Second && sinceHash > 5*time.Second { + self.catchingUp = false + } + + if sinceHash > 10*time.Second && self.ethereum.blockPool.Len() != 0 { + // XXX While this is completely and utterly incorrect, in order to do anything on the test net is to do it this way + // Assume that when fetching hashes timeouts, we are done. + //self.FetchHashes() + self.FetchBlocks() + } } case <-self.quit: break out @@ -761,6 +772,14 @@ func (p *Peer) handleHandshake(msg *ethwire.Msg) { return } + // Self connect detection + pubkey := p.ethereum.KeyManager().PublicKey() + if bytes.Compare(pubkey[1:], pub) == 0 { + p.Stop() + + return + } + usedPub := 0 // This peer is already added to the peerlist so we expect to find a double pubkey at least once eachPeer(p.ethereum.Peers(), func(peer *Peer, e *list.Element) { @@ -779,16 +798,8 @@ func (p *Peer) handleHandshake(msg *ethwire.Msg) { // If this is an inbound connection send an ack back if p.inbound { p.port = uint16(port) - - // Self connect detection - pubkey := p.ethereum.KeyManager().PublicKey() - if bytes.Compare(pubkey, p.pubkey) == 0 { - p.Stop() - - return - } - } + p.SetVersion(clientId) p.versionKnown = true -- cgit From 1fe2d0d0e011b6d6bc6261b69fe955943d9d6e3e Mon Sep 17 00:00:00 2001 From: obscuren Date: Wed, 24 Sep 2014 19:55:28 +0200 Subject: Peers no longer take care of block processing --- peer.go | 34 ++-------------------------------- 1 file changed, 2 insertions(+), 32 deletions(-) (limited to 'peer.go') diff --git a/peer.go b/peer.go index 8224f4831..3bbfb74fa 100644 --- a/peer.go +++ b/peer.go @@ -517,16 +517,9 @@ func (p *Peer) HandleInbound() { break } - blockPool.AddHash(hash) + blockPool.AddHash(hash, p) } - /* - if foundCommonHash || msg.Data.Len() == 0 { - p.FetchBlocks() - } else { - p.FetchHashes() - } - */ if !foundCommonHash && msg.Data.Len() != 0 { p.FetchHashes() } @@ -545,22 +538,6 @@ func (p *Peer) HandleInbound() { p.lastBlockReceived = time.Now() } - - /* - var err error - blockPool.CheckLinkAndProcess(func(block *ethchain.Block) { - err = p.ethereum.StateManager().Process(block, false) - }) - - if err != nil { - peerlogger.Infoln(err) - } else { - // Don't trigger if there's just one block. - if blockPool.Len() != 0 && msg.Data.Len() > 1 { - p.FetchBlocks() - } - } - */ } } } @@ -608,13 +585,6 @@ out: if sinceBlock > 5*time.Second && sinceHash > 5*time.Second { self.catchingUp = false } - - if sinceHash > 10*time.Second && self.ethereum.blockPool.Len() != 0 { - // XXX While this is completely and utterly incorrect, in order to do anything on the test net is to do it this way - // Assume that when fetching hashes timeouts, we are done. - //self.FetchHashes() - //self.FetchBlocks() - } } case <-self.quit: break out @@ -738,7 +708,7 @@ func (self *Peer) handleStatus(msg *ethwire.Msg) { // Compare the total TD with the blockchain TD. If remote is higher // fetch hashes from highest TD node. if self.td.Cmp(self.ethereum.BlockChain().TD) > 0 { - self.ethereum.blockPool.AddHash(self.lastReceivedHash) + self.ethereum.blockPool.AddHash(self.lastReceivedHash, self) self.FetchHashes() } -- cgit