toolbar/precog/monitor/connection.go

   1 package monitor
   2
   3 import (
   4         "net"
   5         "time"
   6
   7         log "github.com/sirupsen/logrus"
   8         "github.com/vapor/p2p"
   9         "github.com/vapor/toolbar/precog/database/orm"
  10 )
  11
  12 func (m *monitor) connectionRoutine() {
  13         // TODO: fix
  14         // ticker := time.NewTicker(time.Duration(m.cfg.CheckFreqMinutes) * time.Minute)
  15         ticker := time.NewTicker(time.Duration(m.cfg.CheckFreqMinutes) * time.Second)
  16         for ; true; <-ticker.C {
  17                 m.Lock()
  18
  19                 if err := m.dialNodes(); err != nil {
  20                         log.WithFields(log.Fields{"err": err}).Error("dialNodes")
  21                 }
  22         }
  23 }
  24
  25 func (m *monitor) dialNodes() error {
  26         log.Info("Start to reconnect to nodes...")
  27         var nodes []*orm.Node
  28         if err := m.db.Model(&orm.Node{}).Find(&nodes).Error; err != nil {
  29                 return err
  30         }
  31
  32         addresses := make([]*p2p.NetAddress, 0)
  33         for i := 0; i < len(nodes); i++ {
  34                 address := p2p.NewNetAddressIPPort(net.ParseIP(nodes[i].IP), nodes[i].Port)
  35                 addresses = append(addresses, address)
  36         }
  37
  38         // connected peers will be skipped in switch.DialPeers()
  39         m.sw.DialPeers(addresses)
  40         log.Info("DialPeers done.")
  41         m.processDialResults()
  42         m.checkStatus()
  43         return nil
  44 }
  45
  46 func (m *monitor) checkStatus() {
  47         for _, peer := range m.sw.GetPeers().List() {
  48                 peer.Start()
  49                 m.peers.AddPeer(peer)
  50         }
  51         log.WithFields(log.Fields{"num": len(m.sw.GetPeers().List()), "peers": m.sw.GetPeers().List()}).Info("connected peers")
  52
  53         for _, peer := range m.sw.GetPeers().List() {
  54                 p := m.peers.GetPeer(peer.ID())
  55                 if p == nil {
  56                         continue
  57                 }
  58
  59                 if err := p.SendStatus(m.chain.BestBlockHeader(), m.chain.LastIrreversibleHeader()); err != nil {
  60                         log.WithFields(log.Fields{"peer": p, "err": err}).Error("SendStatus")
  61                         m.peers.RemovePeer(p.ID())
  62                 }
  63         }
  64
  65         for _, peerInfo := range m.peers.GetPeerInfos() {
  66                 if peerInfo.Height > m.bestHeightSeen {
  67                         m.bestHeightSeen = peerInfo.Height
  68                 }
  69         }
  70         log.WithFields(log.Fields{"bestHeight": m.bestHeightSeen}).Info("peersInfo")
  71         m.processPeerInfos(m.peers.GetPeerInfos())
  72
  73         for _, peer := range m.sw.GetPeers().List() {
  74                 p := m.peers.GetPeer(peer.ID())
  75                 if p == nil {
  76                         continue
  77                 }
  78
  79                 m.peers.RemovePeer(p.ID())
  80         }
  81         log.Info("Disonnect all peers.")
  82
  83         m.Unlock()
  84 }