plugeth/ethereum.go

661 lines
14 KiB
Go
Raw Normal View History

2014-01-23 19:14:01 +00:00
package eth
import (
"container/list"
2014-09-16 14:36:27 +00:00
"encoding/json"
"fmt"
2014-09-24 09:39:17 +00:00
"math/big"
"math/rand"
2014-01-23 19:14:01 +00:00
"net"
2014-09-16 14:36:27 +00:00
"path"
2014-02-02 18:22:39 +00:00
"strconv"
"strings"
2014-02-02 15:15:39 +00:00
"sync"
2014-01-23 19:14:01 +00:00
"sync/atomic"
"time"
2014-07-26 09:24:44 +00:00
2014-10-31 09:59:17 +00:00
"github.com/ethereum/go-ethereum/chain"
2014-10-31 11:37:43 +00:00
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/ethutil"
"github.com/ethereum/go-ethereum/event"
2014-10-31 11:56:05 +00:00
"github.com/ethereum/go-ethereum/logger"
"github.com/ethereum/go-ethereum/rpc"
2014-10-31 13:43:14 +00:00
"github.com/ethereum/go-ethereum/state"
2014-10-31 13:53:42 +00:00
"github.com/ethereum/go-ethereum/wire"
2014-01-23 19:14:01 +00:00
)
2014-07-29 21:33:59 +00:00
const (
seedTextFileUri string = "http://www.ethereum.org/servers.poc3.txt"
2014-10-21 11:25:31 +00:00
seedNodeAddress = "poc-7.ethdev.com:30303"
2014-07-29 21:33:59 +00:00
)
2014-07-11 14:04:09 +00:00
2014-10-31 11:56:05 +00:00
var loggerger = logger.NewLogger("SERV")
2014-01-23 19:14:01 +00:00
func eachPeer(peers *list.List, callback func(*Peer, *list.Element)) {
// Loop thru the peers and close them (if we had them)
for e := peers.Front(); e != nil; e = e.Next() {
2014-09-16 14:36:27 +00:00
callback(e.Value.(*Peer), e)
2014-01-23 19:14:01 +00:00
}
}
const (
processReapingTimeout = 60 // TODO increase
)
type Ethereum struct {
// Channel for shutting down the ethereum
shutdownChan chan bool
2014-02-02 18:22:39 +00:00
quit chan bool
2014-08-11 14:23:38 +00:00
2014-01-23 19:14:01 +00:00
// DB interface
db ethutil.Database
2014-03-05 09:57:32 +00:00
// State manager for processing new blocks and managing the over all states
2014-10-31 09:59:17 +00:00
stateManager *chain.StateManager
2014-01-23 19:14:01 +00:00
// The transaction pool. Transaction can be pushed on this pool
// for later including in the blocks
2014-10-31 09:59:17 +00:00
txPool *chain.TxPool
// The canonical chain
2014-10-31 09:59:17 +00:00
blockChain *chain.ChainManager
// The block pool
blockPool *BlockPool
// Eventer
2014-10-14 00:01:46 +00:00
eventMux event.TypeMux
// Peers
2014-01-23 19:14:01 +00:00
peers *list.List
// Nonce
Nonce uint64
2014-02-01 20:30:54 +00:00
Addr net.Addr
Port string
2014-02-01 20:30:54 +00:00
2014-10-02 15:03:48 +00:00
blacklist [][]byte
2014-02-02 15:15:39 +00:00
peerMut sync.Mutex
// Capabilities for outgoing peers
serverCaps Caps
2014-02-02 18:22:39 +00:00
nat NAT
// Specifies the desired amount of maximum peers
MaxPeers int
2014-03-10 10:53:02 +00:00
Mining bool
listening bool
2014-10-21 11:25:31 +00:00
RpcServer *rpc.JsonRpcServer
2014-10-31 11:37:43 +00:00
keyManager *crypto.KeyManager
2014-10-31 13:53:42 +00:00
clientIdentity wire.ClientIdentity
2014-07-17 20:01:13 +00:00
isUpToDate bool
2014-09-13 22:13:23 +00:00
filterMu sync.RWMutex
filterId int
2014-10-31 09:59:17 +00:00
filters map[int]*chain.Filter
2014-01-23 19:14:01 +00:00
}
2014-10-31 13:53:42 +00:00
func New(db ethutil.Database, clientIdentity wire.ClientIdentity, keyManager *crypto.KeyManager, caps Caps, usePnp bool) (*Ethereum, error) {
var err error
2014-02-02 18:44:47 +00:00
var nat NAT
2014-02-02 18:44:47 +00:00
if usePnp {
nat, err = Discover()
if err != nil {
2014-10-31 11:56:05 +00:00
loggerger.Debugln("UPnP failed", err)
2014-02-02 18:44:47 +00:00
}
2014-02-02 18:22:39 +00:00
}
bootstrapDb(db)
2014-01-23 19:14:01 +00:00
ethutil.Config.Db = db
nonce, _ := ethutil.RandomUint64()
ethereum := &Ethereum{
shutdownChan: make(chan bool),
quit: make(chan bool),
db: db,
peers: list.New(),
Nonce: nonce,
serverCaps: caps,
nat: nat,
keyManager: keyManager,
clientIdentity: clientIdentity,
2014-07-17 20:01:13 +00:00
isUpToDate: true,
2014-10-31 09:59:17 +00:00
filters: make(map[int]*chain.Filter),
2014-01-23 19:14:01 +00:00
}
2014-03-10 10:53:02 +00:00
ethereum.blockPool = NewBlockPool(ethereum)
2014-10-31 09:59:17 +00:00
ethereum.txPool = chain.NewTxPool(ethereum)
ethereum.blockChain = chain.NewChainManager(ethereum)
ethereum.stateManager = chain.NewStateManager(ethereum)
2014-01-23 19:14:01 +00:00
// Start the tx pool
ethereum.txPool.Start()
2014-01-23 19:14:01 +00:00
return ethereum, nil
}
2014-10-31 11:37:43 +00:00
func (s *Ethereum) KeyManager() *crypto.KeyManager {
return s.keyManager
}
2014-10-31 13:53:42 +00:00
func (s *Ethereum) ClientIdentity() wire.ClientIdentity {
return s.clientIdentity
}
2014-10-31 09:59:17 +00:00
func (s *Ethereum) ChainManager() *chain.ChainManager {
return s.blockChain
}
2014-10-31 09:59:17 +00:00
func (s *Ethereum) StateManager() *chain.StateManager {
2014-03-05 09:57:32 +00:00
return s.stateManager
}
2014-10-31 09:59:17 +00:00
func (s *Ethereum) TxPool() *chain.TxPool {
return s.txPool
}
2014-09-24 17:56:21 +00:00
func (s *Ethereum) BlockPool() *BlockPool {
return s.blockPool
}
func (s *Ethereum) EventMux() *event.TypeMux {
2014-10-14 00:01:46 +00:00
return &s.eventMux
}
2014-08-11 14:23:38 +00:00
func (self *Ethereum) Db() ethutil.Database {
return self.db
}
func (s *Ethereum) ServerCaps() Caps {
return s.serverCaps
}
func (s *Ethereum) IsMining() bool {
return s.Mining
}
func (s *Ethereum) PeerCount() int {
return s.peers.Len()
}
func (s *Ethereum) IsUpToDate() bool {
upToDate := true
eachPeer(s.peers, func(peer *Peer, e *list.Element) {
if atomic.LoadInt32(&peer.connected) == 1 {
2014-07-18 09:57:44 +00:00
if peer.catchingUp == true && peer.versionKnown {
upToDate = false
}
}
})
return upToDate
}
func (s *Ethereum) PushPeer(peer *Peer) {
s.peers.PushBack(peer)
}
func (s *Ethereum) IsListening() bool {
return s.listening
}
2014-09-24 09:39:17 +00:00
func (s *Ethereum) HighestTDPeer() (td *big.Int) {
td = big.NewInt(0)
eachPeer(s.peers, func(p *Peer, v *list.Element) {
if p.td.Cmp(td) > 0 {
td = p.td
}
})
return
}
2014-10-02 15:03:48 +00:00
func (self *Ethereum) BlacklistPeer(peer *Peer) {
self.blacklist = append(self.blacklist, peer.pubkey)
}
2014-01-23 19:14:01 +00:00
func (s *Ethereum) AddPeer(conn net.Conn) {
peer := NewPeer(conn, s, true)
if peer != nil {
if s.peers.Len() < s.MaxPeers {
peer.Start()
} else {
2014-10-31 11:56:05 +00:00
loggerger.Debugf("Max connected peers reached. Not adding incoming peer.")
}
2014-01-23 19:14:01 +00:00
}
}
func (s *Ethereum) ProcessPeerList(addrs []string) {
for _, addr := range addrs {
// TODO Probably requires some sanity checks
s.ConnectToPeer(addr)
}
}
func (s *Ethereum) ConnectToPeer(addr string) error {
2014-02-18 00:34:06 +00:00
if s.peers.Len() < s.MaxPeers {
var alreadyConnected bool
ahost, _, _ := net.SplitHostPort(addr)
var chost string
ips, err := net.LookupIP(ahost)
if err != nil {
return err
} else {
// If more then one ip is available try stripping away the ipv6 ones
if len(ips) > 1 {
var ipsv4 []net.IP
// For now remove the ipv6 addresses
for _, ip := range ips {
if strings.Contains(ip.String(), "::") {
continue
} else {
ipsv4 = append(ipsv4, ip)
}
}
if len(ipsv4) == 0 {
return fmt.Errorf("[SERV] No IPV4 addresses available for hostname")
}
// Pick a random ipv4 address, simulating round-robin DNS.
rand.Seed(time.Now().UTC().UnixNano())
i := rand.Intn(len(ipsv4))
chost = ipsv4[i].String()
} else {
if len(ips) == 0 {
return fmt.Errorf("[SERV] No IPs resolved for the given hostname")
return nil
}
chost = ips[0].String()
}
}
2014-02-18 00:34:06 +00:00
eachPeer(s.peers, func(p *Peer, v *list.Element) {
if p.conn == nil {
return
}
phost, _, _ := net.SplitHostPort(p.conn.RemoteAddr().String())
if phost == chost {
2014-02-18 00:34:06 +00:00
alreadyConnected = true
2014-10-31 11:56:05 +00:00
//loggerger.Debugf("Peer %s already added.\n", chost)
2014-02-18 00:34:06 +00:00
return
}
})
if alreadyConnected {
return nil
}
NewOutboundPeer(addr, s, s.serverCaps)
2014-02-18 00:34:06 +00:00
}
2014-01-23 19:14:01 +00:00
return nil
}
func (s *Ethereum) OutboundPeers() []*Peer {
// Create a new peer slice with at least the length of the total peers
outboundPeers := make([]*Peer, s.peers.Len())
length := 0
eachPeer(s.peers, func(p *Peer, e *list.Element) {
2014-01-31 10:18:10 +00:00
if !p.inbound && p.conn != nil {
2014-01-23 19:14:01 +00:00
outboundPeers[length] = p
length++
}
})
return outboundPeers[:length]
}
func (s *Ethereum) InboundPeers() []*Peer {
// Create a new peer slice with at least the length of the total peers
inboundPeers := make([]*Peer, s.peers.Len())
length := 0
eachPeer(s.peers, func(p *Peer, e *list.Element) {
if p.inbound {
inboundPeers[length] = p
length++
}
})
return inboundPeers[:length]
}
func (s *Ethereum) InOutPeers() []*Peer {
2014-02-02 15:15:39 +00:00
// Reap the dead peers first
s.reapPeers()
// Create a new peer slice with at least the length of the total peers
inboundPeers := make([]*Peer, s.peers.Len())
length := 0
eachPeer(s.peers, func(p *Peer, e *list.Element) {
2014-02-02 15:15:39 +00:00
// Only return peers with an actual ip
if len(p.host) > 0 {
inboundPeers[length] = p
length++
}
})
return inboundPeers[:length]
}
2014-10-31 13:53:42 +00:00
func (s *Ethereum) Broadcast(msgType wire.MsgType, data []interface{}) {
msg := wire.NewMessage(msgType, data)
2014-02-02 15:15:39 +00:00
s.BroadcastMsg(msg)
}
2014-10-31 13:53:42 +00:00
func (s *Ethereum) BroadcastMsg(msg *wire.Msg) {
2014-01-23 19:14:01 +00:00
eachPeer(s.peers, func(p *Peer, e *list.Element) {
p.QueueMessage(msg)
2014-01-23 19:14:01 +00:00
})
}
func (s *Ethereum) Peers() *list.List {
return s.peers
}
2014-02-02 15:15:39 +00:00
func (s *Ethereum) reapPeers() {
eachPeer(s.peers, func(p *Peer, e *list.Element) {
if atomic.LoadInt32(&p.disconnect) == 1 || (p.inbound && (time.Now().Unix()-p.lastPong) > int64(5*time.Minute)) {
s.removePeerElement(e)
}
})
}
func (s *Ethereum) removePeerElement(e *list.Element) {
2014-02-02 15:15:39 +00:00
s.peerMut.Lock()
defer s.peerMut.Unlock()
s.peers.Remove(e)
s.eventMux.Post(PeerListEvent{s.peers})
}
func (s *Ethereum) RemovePeer(p *Peer) {
eachPeer(s.peers, func(peer *Peer, e *list.Element) {
if peer == p {
s.removePeerElement(e)
2014-02-02 15:15:39 +00:00
}
})
}
2014-01-23 19:14:01 +00:00
2014-10-08 10:29:49 +00:00
func (s *Ethereum) reapDeadPeerHandler() {
2014-02-02 15:15:39 +00:00
reapTimer := time.NewTicker(processReapingTimeout * time.Second)
for {
select {
case <-reapTimer.C:
s.reapPeers()
}
2014-01-23 19:14:01 +00:00
}
}
// Start the ethereum
2014-05-09 14:09:28 +00:00
func (s *Ethereum) Start(seed bool) {
2014-09-24 19:13:28 +00:00
s.blockPool.Start()
s.stateManager.Start()
2014-01-27 21:13:46 +00:00
// Bind to addr and port
ln, err := net.Listen("tcp", ":"+s.Port)
2014-01-23 19:14:01 +00:00
if err != nil {
2014-10-31 11:56:05 +00:00
loggerger.Warnf("Port %s in use. Connection listening disabled. Acting as client", s.Port)
s.listening = false
2014-01-23 19:14:01 +00:00
} else {
s.listening = true
2014-01-23 19:14:01 +00:00
// Starting accepting connections
2014-10-31 11:56:05 +00:00
loggerger.Infoln("Ready and accepting connections")
2014-02-02 15:15:39 +00:00
// Start the peer handler
go s.peerHandler(ln)
2014-01-23 19:14:01 +00:00
}
2014-02-02 18:44:47 +00:00
if s.nat != nil {
go s.upnpUpdateThread()
}
2014-02-02 18:22:39 +00:00
2014-01-23 19:14:01 +00:00
// Start the reaping processes
2014-10-08 10:29:49 +00:00
go s.reapDeadPeerHandler()
2014-07-17 20:01:13 +00:00
go s.update()
2014-09-13 22:13:23 +00:00
go s.filterLoop()
2014-01-23 19:14:01 +00:00
2014-05-09 14:09:28 +00:00
if seed {
s.Seed()
}
2014-10-31 11:56:05 +00:00
loggerger.Infoln("Server started")
2014-05-09 14:09:28 +00:00
}
func (s *Ethereum) Seed() {
2014-10-08 10:06:39 +00:00
// Sorry Py person. I must blacklist. you perform badly
s.blacklist = append(s.blacklist, ethutil.Hex2Bytes("64656330303561383532336435376331616537643864663236623336313863373537353163636634333530626263396330346237336262623931383064393031"))
2014-09-17 13:57:44 +00:00
ips := PastPeers()
2014-09-16 14:36:27 +00:00
if len(ips) > 0 {
for _, ip := range ips {
2014-10-31 11:56:05 +00:00
loggerger.Infoln("Connecting to previous peer ", ip)
2014-09-16 14:36:27 +00:00
s.ConnectToPeer(ip)
}
} else {
2014-10-31 11:56:05 +00:00
loggerger.Debugln("Retrieving seed nodes")
2014-09-16 14:36:27 +00:00
// Eth-Go Bootstrapping
ips, er := net.LookupIP("seed.bysh.me")
if er == nil {
peers := []string{}
for _, ip := range ips {
node := fmt.Sprintf("%s:%d", ip.String(), 30303)
2014-10-31 11:56:05 +00:00
loggerger.Debugln("Found DNS Go Peer:", node)
2014-09-16 14:36:27 +00:00
peers = append(peers, node)
}
s.ProcessPeerList(peers)
}
2014-09-16 14:36:27 +00:00
// Official DNS Bootstrapping
_, nodes, err := net.LookupSRV("eth", "tcp", "ethereum.org")
if err == nil {
peers := []string{}
// Iterate SRV nodes
for _, n := range nodes {
target := n.Target
port := strconv.Itoa(int(n.Port))
// Resolve target to ip (Go returns list, so may resolve to multiple ips?)
addr, err := net.LookupHost(target)
if err == nil {
for _, a := range addr {
// Build string out of SRV port and Resolved IP
peer := net.JoinHostPort(a, port)
2014-10-31 11:56:05 +00:00
loggerger.Debugln("Found DNS Bootstrap Peer:", peer)
2014-09-16 14:36:27 +00:00
peers = append(peers, peer)
}
} else {
2014-10-31 11:56:05 +00:00
loggerger.Debugln("Couldn't resolve :", target)
2014-03-17 09:37:37 +00:00
}
}
2014-09-16 14:36:27 +00:00
// Connect to Peer list
s.ProcessPeerList(peers)
2014-03-17 09:37:37 +00:00
}
2014-07-29 21:33:59 +00:00
2014-09-16 14:36:27 +00:00
// XXX tmp
s.ConnectToPeer(seedNodeAddress)
}
2014-01-23 19:14:01 +00:00
}
2014-02-02 15:15:39 +00:00
func (s *Ethereum) peerHandler(listener net.Listener) {
for {
conn, err := listener.Accept()
if err != nil {
2014-10-31 11:56:05 +00:00
loggerger.Debugln(err)
2014-02-02 15:15:39 +00:00
continue
}
go s.AddPeer(conn)
}
}
2014-01-23 19:14:01 +00:00
func (s *Ethereum) Stop() {
// Stop eventMux first, it will close all subscriptions.
s.eventMux.Stop()
2014-01-23 19:14:01 +00:00
// Close the database
defer s.db.Close()
2014-09-16 14:36:27 +00:00
var ips []string
eachPeer(s.peers, func(p *Peer, e *list.Element) {
ips = append(ips, p.conn.RemoteAddr().String())
})
2014-09-17 13:57:44 +00:00
if len(ips) > 0 {
d, _ := json.MarshalIndent(ips, "", " ")
ethutil.WriteFile(path.Join(ethutil.Config.ExecPath, "known_peers.json"), d)
}
2014-09-16 14:36:27 +00:00
2014-01-23 19:14:01 +00:00
eachPeer(s.peers, func(p *Peer, e *list.Element) {
p.Stop()
})
2014-02-02 18:22:39 +00:00
close(s.quit)
if s.RpcServer != nil {
s.RpcServer.Stop()
}
s.txPool.Stop()
2014-03-05 09:57:32 +00:00
s.stateManager.Stop()
2014-09-24 17:56:21 +00:00
s.blockPool.Stop()
2014-10-31 11:56:05 +00:00
loggerger.Infoln("Server stopped")
2014-02-28 15:45:29 +00:00
close(s.shutdownChan)
2014-01-23 19:14:01 +00:00
}
// This function will wait for a shutdown and resumes main thread execution
func (s *Ethereum) WaitForShutdown() {
<-s.shutdownChan
}
2014-02-02 18:22:39 +00:00
func (s *Ethereum) upnpUpdateThread() {
// Go off immediately to prevent code duplication, thereafter we renew
// lease every 15 minutes.
2014-05-01 20:14:20 +00:00
timer := time.NewTimer(5 * time.Minute)
lport, _ := strconv.ParseInt(s.Port, 10, 16)
2014-02-02 18:22:39 +00:00
first := true
out:
for {
select {
case <-timer.C:
2014-02-02 18:44:47 +00:00
var err error
_, err = s.nat.AddPortMapping("TCP", int(lport), int(lport), "eth listen port", 20*60)
2014-02-02 18:22:39 +00:00
if err != nil {
2014-10-31 11:56:05 +00:00
loggerger.Debugln("can't add UPnP port mapping:", err)
2014-02-02 18:22:39 +00:00
break out
}
if first && err == nil {
2014-02-02 18:44:47 +00:00
_, err = s.nat.GetExternalAddress()
2014-02-02 18:22:39 +00:00
if err != nil {
2014-10-31 11:56:05 +00:00
loggerger.Debugln("UPnP can't get external address:", err)
2014-02-02 18:22:39 +00:00
continue out
}
first = false
}
timer.Reset(time.Minute * 15)
case <-s.quit:
break out
}
}
timer.Stop()
if err := s.nat.DeletePortMapping("TCP", int(lport), int(lport)); err != nil {
2014-10-31 11:56:05 +00:00
loggerger.Debugln("unable to remove UPnP port mapping:", err)
2014-02-02 18:22:39 +00:00
} else {
2014-10-31 11:56:05 +00:00
loggerger.Debugln("succesfully disestablished UPnP port mapping")
2014-02-02 18:22:39 +00:00
}
}
2014-07-17 20:01:13 +00:00
func (self *Ethereum) update() {
upToDateTimer := time.NewTicker(1 * time.Second)
out:
for {
select {
case <-upToDateTimer.C:
if self.IsUpToDate() && !self.isUpToDate {
self.eventMux.Post(ChainSyncEvent{false})
2014-07-17 20:01:13 +00:00
self.isUpToDate = true
} else if !self.IsUpToDate() && self.isUpToDate {
self.eventMux.Post(ChainSyncEvent{true})
2014-07-17 20:01:13 +00:00
self.isUpToDate = false
}
case <-self.quit:
break out
}
}
}
// InstallFilter adds filter for blockchain events.
// The filter's callbacks will run for matching blocks and messages.
// The filter should not be modified after it has been installed.
2014-10-31 09:59:17 +00:00
func (self *Ethereum) InstallFilter(filter *chain.Filter) (id int) {
self.filterMu.Lock()
id = self.filterId
self.filters[id] = filter
self.filterId++
self.filterMu.Unlock()
return id
2014-09-13 22:13:23 +00:00
}
func (self *Ethereum) UninstallFilter(id int) {
self.filterMu.Lock()
2014-09-13 22:13:23 +00:00
delete(self.filters, id)
self.filterMu.Unlock()
2014-09-13 22:13:23 +00:00
}
// GetFilter retrieves a filter installed using InstallFilter.
// The filter may not be modified.
2014-10-31 09:59:17 +00:00
func (self *Ethereum) GetFilter(id int) *chain.Filter {
self.filterMu.RLock()
defer self.filterMu.RUnlock()
2014-09-13 22:13:23 +00:00
return self.filters[id]
}
func (self *Ethereum) filterLoop() {
// Subscribe to events
2014-10-31 13:43:14 +00:00
events := self.eventMux.Subscribe(chain.NewBlockEvent{}, state.Messages(nil))
for event := range events.Chan() {
switch event := event.(type) {
2014-10-31 09:59:17 +00:00
case chain.NewBlockEvent:
self.filterMu.RLock()
for _, filter := range self.filters {
if filter.BlockCallback != nil {
filter.BlockCallback(event.Block)
2014-09-13 22:13:23 +00:00
}
}
self.filterMu.RUnlock()
2014-10-31 13:43:14 +00:00
case state.Messages:
self.filterMu.RLock()
for _, filter := range self.filters {
if filter.MessageCallback != nil {
msgs := filter.FilterMessages(event)
if len(msgs) > 0 {
filter.MessageCallback(msgs)
2014-09-13 22:13:23 +00:00
}
}
}
self.filterMu.RUnlock()
2014-09-13 22:13:23 +00:00
}
}
}
func bootstrapDb(db ethutil.Database) {
d, _ := db.Get([]byte("ProtocolVersion"))
protov := ethutil.NewValue(d).Uint()
if protov == 0 {
db.Put([]byte("ProtocolVersion"), ethutil.NewValue(ProtocolVersion).Bytes())
}
}
2014-09-17 13:57:44 +00:00
func PastPeers() []string {
var ips []string
data, _ := ethutil.ReadAllFile(path.Join(ethutil.Config.ExecPath, "known_peers.json"))
json.Unmarshal([]byte(data), &ips)
return ips
}