From 72deabeb422b4ffacfa79c12d5baf2251dd3e74c Mon Sep 17 00:00:00 2001 From: zelig Date: Tue, 6 Jan 2015 18:20:07 +0000 Subject: [PATCH] simplify server logic - get rid of dialLoop entirely - get rid of peerConnect channel in favour of simple function call - server handles the peerSelector hook for prospective outbound peers - no need for NoDial anymore really - changed how peer is initialised --- p2p/server.go | 160 +++++++++++++++++++------------------------------- 1 file changed, 59 insertions(+), 101 deletions(-) diff --git a/p2p/server.go b/p2p/server.go index dff4738ad5..ec09647011 100644 --- a/p2p/server.go +++ b/p2p/server.go @@ -67,7 +67,7 @@ type Server struct { // Hook for testing. This is useful because we can inhibit // the whole protocol stack. - newPeerFunc peerFunc + // newPeerFunc peerFunc lock sync.RWMutex running bool @@ -79,7 +79,6 @@ type Server struct { quit chan struct{} wg sync.WaitGroup - peerConnect chan *peerAddr peerDisconnect chan *Peer } @@ -95,7 +94,7 @@ type NAT interface { type peerFunc func(srv *Server, c net.Conn, dialAddr *peerAddr) *Peer -// Peers returns all connected peers. +// Peers returns all currently connected peers. func (srv *Server) Peers() (peers []*Peer) { srv.lock.RLock() defer srv.lock.RUnlock() @@ -107,33 +106,18 @@ func (srv *Server) Peers() (peers []*Peer) { return } -// GetActivePeers returns addresses of all connected peers. -func (srv *Server) GetActivePeers() (peers []*peerAddr) { - for _, peer := range srv.Peers() { - if peer != nil { - peer.infolock.Lock() - addr := peer.listenAddr - peer.infolock.Unlock() - // filter out this peer and peers that are not listening or - // have not completed the handshake. - // TODO: track previously sent peers and exclude them as well. - if addr == nil { - continue - } - peers = append(peers, addr) - } - } - return peers -} - // GetPeers returns addresses near target if given or supported by the client // or falls back to all actively connected peers -func (srv *Server) GetPeers(target ...[]byte) (peers []*peerAddr) { +func (srv *Server) GetPeers(target ...[]byte) []*peerAddr { if len(target) == 1 { // delegate to selector - srv.PeerSelector.GetPeers(target[0]) - return peers + return ActiveAddresses(srv.PeerSelector.GetPeers(target[0])...) } else { - return srv.GetActivePeers() + // in fact it is not clear why the selector would not want to reply to this case as well + var peers []peerInfo + for _, peer := range srv.Peers() { + peers = append(peers, peer) + } + return ActiveAddresses(peers...) } } @@ -152,7 +136,7 @@ func (srv *Server) SuggestPeer(addr string, pubkey []byte) error { srvlog.Errorf("couldn't resolve %s:", addr, err) return err } - peerAddr := &peerAddr{IP: netaddr.IP, Port: uint64(netaddr.Port), Pubkey: pubkey} + peerAddr := &peerAddr{netaddr.IP, uint64(netaddr.Port), pubkey} return srv.AddPeer(peerAddr) } @@ -161,15 +145,43 @@ func (srv *Server) SuggestPeer(addr string, pubkey []byte) error { // to decide if it is a worthwhile connection func (srv *Server) AddPeer(addr *peerAddr) (err error) { // need to look up nodeID first - peer := &peerRecord{addr: addr} - if err = srv.PeerSelector.AddPeer(peer); err == nil { - srvlog.Infof("peer %v accepted by peer selection", peer) + peer := &Peer{ + dialAddr: addr, + lastActiveC: make(chan time.Time), + lastActive: time.Now().Add(-24 * time.Hour), + } + if srv.PeerSelector.AddPeer(peer) { + srvlog.Infof("peer %v accepted by peer selection", addr) + err = srv.dialPeer(peer) } else { - srvlog.Infof("peer %v rejected by peer selection", peer) + srvlog.Infof("peer %v rejected by peer selection", addr) } return } +func (srv *Server) dialPeer(peer *Peer) (err error) { + timeout := time.After(5 * time.Second) + select { + case <-timeout: + err = fmt.Errorf("Too many connections. No slot available") + case slot := <-srv.peerSlots: // there is a slot available + srvlog.Debugf("Dialing %v (slot %d)\n", peer.dialAddr, slot) + conn, dialErr := srv.Dialer.Dial(peer.dialAddr.Network(), peer.dialAddr.String()) + if dialErr != nil { + err = fmt.Errorf("Dial error: %v", dialErr) + srvlog.Errorln(err) + srv.peerSlots <- slot + return + } + peer.slot = slot + peer.connect(srv, conn) + srv.wg.Add(1) + go srv.addPeer(peer) + } + return + +} + // Broadcast sends an RLP-encoded message to all connected peers. // This method is deprecated and will be removed later. func (srv *Server) Broadcast(protocol string, code uint64, data ...interface{}) { @@ -211,11 +223,10 @@ func (srv *Server) Start() (err error) { srv.quit = make(chan struct{}) srv.peers = make([]*Peer, srv.MaxPeers) srv.peerSlots = make(chan int, srv.MaxPeers) - srv.peerConnect = make(chan *peerAddr, outboundAddressPoolSize) srv.peerDisconnect = make(chan *Peer) - if srv.newPeerFunc == nil { - srv.newPeerFunc = newServerPeer - } + // if srv.newPeerFunc == nil { + // srv.newPeerFunc = newServerPeer + // } if srv.Blacklist == nil { srv.Blacklist = NewBlacklist() } @@ -228,10 +239,10 @@ func (srv *Server) Start() (err error) { return err } } - if !srv.NoDial { - srv.wg.Add(1) - go srv.dialLoop() - } + // if !srv.NoDial { + // srv.wg.Add(1) + // go srv.dialLoop() + // } if srv.NoDial && srv.ListenAddr == "" { srvlog.Warnln("I will be kind-of useless, neither dialing nor listening.") } @@ -319,8 +330,10 @@ func (srv *Server) listenLoop() { srv.peerSlots <- slot return } - srvlog.Debugf("Accepted conn %v (slot %d)\n", conn.RemoteAddr(), slot) - srv.addPeer(conn, nil, slot) + srvlog.Debugf("Accepted conn %v (slot %d) - peer selector check after handshake", conn.RemoteAddr(), slot) + peer := &Peer{slot: slot} + peer.connect(srv, conn) + srv.addPeer(peer) case <-srv.quit: return } @@ -366,77 +379,22 @@ func (srv *Server) removePortMapping(port int) { srv.NAT.DeletePortMapping("tcp", port, port) } -func (srv *Server) dialLoop() { - defer srv.wg.Done() - var ( - suggest chan *peerAddr - slot *int - slots = srv.peerSlots - ) - for { - select { - case i := <-slots: - // we need a peer in slot i, slot reserved - slot = &i - // now we can watch for candidate peers in the next loop - suggest = srv.peerConnect - // do not consume more until candidate peer is found - slots = nil - - case desc := <-suggest: - // candidate peer found, will dial out asyncronously - // if connection fails slot will be released - srvlog.Infof("dial %v (%v)", desc, *slot) - go srv.dialPeer(desc, *slot) - // we can watch if more peers needed in the next loop - slots = srv.peerSlots - // until then we dont care about candidate peers - suggest = nil - - case <-srv.quit: - // give back the currently reserved slot - if slot != nil { - srv.peerSlots <- *slot - } - return - } - } -} - -// connect to peer via dial out -func (srv *Server) dialPeer(desc *peerAddr, slot int) { - srvlog.Debugf("Dialing %v (slot %d)\n", desc, slot) - conn, err := srv.Dialer.Dial(desc.Network(), desc.String()) - if err != nil { - srvlog.Errorf("Dial error: %v", err) - srv.peerSlots <- slot - return - } - go srv.addPeer(conn, desc, slot) -} - // creates the new peer object and inserts it into its slot -func (srv *Server) addPeer(conn net.Conn, desc *peerAddr, slot int) *Peer { +func (srv *Server) addPeer(peer *Peer) { srv.lock.Lock() defer srv.lock.Unlock() - if !srv.running { - conn.Close() - srv.peerSlots <- slot // release slot - return nil - } - peer := srv.newPeerFunc(srv, conn, desc) - peer.slot = slot - srv.peers[slot] = peer + srvlog.Debugf("Add peer %v (slot %v)\n", peer, peer.slot) + srv.peers[peer.slot] = peer srv.peerCount++ go func() { peer.loop(); srv.peerDisconnect <- peer }() - return peer + return } // removes peer: sending disconnect msg, stop peer, remove rom list/table, release slot func (srv *Server) removePeer(peer *Peer) { srv.lock.Lock() defer srv.lock.Unlock() - srvlog.Debugf("Removing %v (slot %v)\n", peer, peer.slot) + srvlog.Debugf("Remove peer %v (slot %v)\n", peer, peer.slot) if srv.peers[peer.slot] != peer { srvlog.Warnln("Invalid peer to remove:", peer) return