p2p/discover: add simulation setup

p2p/discover/net.go

@@ -100,10 +100,15 @@ type timeoutEvent struct {
 
 func newNetwork(conn transport, ourPubkey ecdsa.PublicKey, natm nat.Interface, dbPath string) (*Network, error) {
 	ourID := PubkeyID(&ourPubkey)
-	db, err := newNodeDB(dbPath, Version, ourID)
+
-	if err != nil {
+	var db *nodeDB
+	if dbPath != "<no database>" {
+		var err error
+		if db, err = newNodeDB(dbPath, Version, ourID); err != nil {
 			return nil, err
 		}
+	}
+
 	net := &Network{
 		db:            db,
 		conn:          conn,
@@ -112,7 +117,7 @@ func newNetwork(conn transport, ourPubkey ecdsa.PublicKey, natm nat.Interface, d
 		refreshResp:   make(chan (<-chan struct{})),
 		closed:        make(chan struct{}),
 		closeReq:      make(chan struct{}),
-		read:          make(chan ingressPacket),
+		read:          make(chan ingressPacket, 100),
 		timeout:       make(chan timeoutEvent),
 		timeoutTimers: make(map[timeoutEvent]*time.Timer),
 		tableOpReq:    make(chan func()),
@@ -291,7 +296,6 @@ loop:
 				glog.Infof("<<< (%d) %v from %x@%v: %v -> %v (%v)",
 					net.tab.count, pkt.ev, pkt.remoteID[:8], pkt.remoteAddr, prestate, n.state, status)
 			}
-
 			// TODO: persist state if n.state goes >= known, delete if it goes <= known
 
 		// State transition timeouts.
@@ -356,7 +360,9 @@ loop:
 	for _, timer := range net.timeoutTimers {
 		timer.Stop()
 	}
+	if net.db != nil {
 		net.db.close()
+	}
 	close(net.closed)
 }
 
@@ -364,7 +370,10 @@ loop:
 // and can modify Node, Table and Network at any time without locking.
 
 func (net *Network) refresh(done chan<- struct{}) {
-	seeds := net.db.querySeeds(seedCount, seedMaxAge)
+	var seeds []*Node
+	if net.db != nil {
+		seeds = net.db.querySeeds(seedCount, seedMaxAge)
+	}
 	if len(seeds) == 0 {
 		seeds = net.nursery
 	}
@@ -375,8 +384,13 @@ func (net *Network) refresh(done chan<- struct{}) {
 	}
 	for _, n := range seeds {
 		if glog.V(logger.Debug) {
-			age := time.Since(net.db.lastPong(n.ID))
+			var age string
-			glog.Infof("seed node (age %v): %v", age, n)
+			if net.db != nil {
+				age = time.Since(net.db.lastPong(n.ID)).String()
+			} else {
+				age = "unknown"
+			}
+			glog.Infof("seed node (age %s): %v", age, n)
 		}
 		n = net.internNodeFromDB(n)
 		if n.state == unknown {
@@ -699,8 +713,10 @@ func (net *Network) handle(n *Node, ev nodeEvent, pkt *ingressPacket) error {
 		// is already running. We do this here instead of somewhere else
 		// so that the search for seed nodes also considers older nodes
 		// that would otherwise be removed by the expirer.
+		if net.db != nil {
 			net.db.ensureExpirer()
 		}
+	}
 	next, err := n.state.handle(net, n, ev, pkt)
 	net.transition(n, next)
 	return err

p2p/discover/net_test.go

@@ -24,13 +24,9 @@ import (
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/crypto"
-	"github.com/ethereum/go-ethereum/logger/glog"
 )
 
 func TestNetwork_Lookup(t *testing.T) {
-	glog.SetV(6)
-	glog.SetToStderr(true)
-
 	key, _ := crypto.GenerateKey()
 	network, err := newNetwork(lookupTestnet, key.PublicKey, nil, "")
 	if err != nil {

p2p/discover/sim_run_test.go

@@ -0,0 +1,126 @@
+// Copyright 2016 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package discover
+
+import (
+	"bufio"
+	"bytes"
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"os/exec"
+	"runtime"
+	"strings"
+	"testing"
+)
+
+func getnacl() (string, error) {
+	switch runtime.GOARCH {
+	case "amd64":
+		_, err := exec.LookPath("sel_ldr_x86_64")
+		return "amd64p32", err
+	case "i386":
+		_, err := exec.LookPath("sel_ldr_i386")
+		return "i386", err
+	default:
+		return "", errors.New("nacl is not supported on " + runtime.GOARCH)
+	}
+}
+
+// runWithPlaygroundTime executes the caller
+// in the NaCl sandbox with faketime enabled.
+//
+// This function must be called from a Test* function
+// and the caller must skip the actual test when isHost is true.
+func runWithPlaygroundTime(t *testing.T) (isHost bool) {
+	if runtime.GOOS == "nacl" {
+		return false
+	}
+
+	// Get the caller.
+	callerPC, _, _, ok := runtime.Caller(1)
+	if !ok {
+		panic("can't get caller")
+	}
+	callerFunc := runtime.FuncForPC(callerPC)
+	if callerFunc == nil {
+		panic("can't get caller")
+	}
+	callerName := callerFunc.Name()[strings.LastIndexByte(callerFunc.Name(), '.')+1:]
+	if !strings.HasPrefix(callerName, "Test") {
+		panic("must be called from witin a Test* function")
+	}
+	testPattern := "^" + callerName + "$"
+
+	// Unfortunately runtime.faketime (playground time mode) only works on NaCl. The NaCl
+	// SDK must be installed and linked into PATH for this to work.
+	arch, err := getnacl()
+	if err != nil {
+		t.Skip(err)
+	}
+
+	// Compile and run the calling test using NaCl.
+	// The extra tag ensures that the TestMain function in sim_main_test.go is used.
+	cmd := exec.Command("go", "test", "-v", "-tags", "faketime_simulation", "-run", testPattern, ".")
+	cmd.Env = append([]string{"GOOS=nacl", "GOARCH=" + arch}, os.Environ()...)
+	stdout, _ := cmd.StdoutPipe()
+	stderr, _ := cmd.StderrPipe()
+	go skipPlaygroundOutputHeaders(os.Stdout, stdout)
+	go skipPlaygroundOutputHeaders(os.Stderr, stderr)
+	if err := cmd.Run(); err != nil {
+		t.Error(err)
+	}
+
+	// Ensure that the test function doesn't run in the (non-NaCl) host process.
+	return true
+}
+
+func skipPlaygroundOutputHeaders(out io.Writer, in io.Reader) {
+	// Additional output can be printed without the headers
+	// before the NaCl binary starts running (e.g. compiler error messages).
+	bufin := bufio.NewReader(in)
+	output, err := bufin.ReadBytes(0)
+	output = bytes.TrimSuffix(output, []byte{0})
+	if len(output) > 0 {
+		out.Write(output)
+	}
+	if err != nil {
+		return
+	}
+	bufin.UnreadByte()
+
+	// Playback header: 0 0 P B <8-byte time> <4-byte data length>
+	head := make([]byte, 4+8+4)
+	for {
+		if _, err := io.ReadFull(bufin, head); err != nil {
+			if err != io.EOF {
+				fmt.Fprintln(out, "read error:", err)
+			}
+			return
+		}
+		if !bytes.HasPrefix(head, []byte{0x00, 0x00, 'P', 'B'}) {
+			fmt.Fprintf(out, "expected playback header, got %q\n", head)
+			io.Copy(out, bufin)
+			return
+		}
+		// Copy data until next header.
+		size := binary.BigEndian.Uint32(head[12:])
+		io.CopyN(out, bufin, int64(size))
+	}
+}

p2p/discover/sim_test.go

@@ -0,0 +1,224 @@
+// Copyright 2016 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package discover
+
+import (
+	"encoding/binary"
+	"fmt"
+	"net"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+)
+
+func TestSimBasic(t *testing.T) {
+	if runWithPlaygroundTime(t) {
+		return
+	}
+
+	sim := newSimulation()
+	bootnode := sim.launchNode()
+
+	// A new node joins every 10s.
+	launcher := time.NewTicker(10 * time.Second)
+	go func() {
+		for range launcher.C {
+			net := sim.launchNode()
+			if err := net.SetFallbackNodes([]*Node{bootnode.Self()}); err != nil {
+				panic(err)
+			}
+			fmt.Printf("launched @ %v: %x\n", time.Now(), net.Self().ID[:16])
+		}
+	}()
+
+	// Print statistics every hour.
+	statsticker := time.NewTicker(1 * time.Hour)
+	go func() {
+		for range statsticker.C {
+			sim.printStats()
+		}
+	}()
+
+	time.Sleep(2 * 24 * time.Hour)
+	launcher.Stop()
+	sim.shutdown()
+	sim.printStats()
+}
+
+type simulation struct {
+	mu      sync.Mutex
+	nodes   map[NodeID]*Network
+	nodectr uint32
+}
+
+func newSimulation() *simulation {
+	return &simulation{nodes: make(map[NodeID]*Network)}
+}
+
+func (s *simulation) shutdown() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	for _, n := range s.nodes {
+		n.Close()
+	}
+}
+
+func (s *simulation) printStats() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	fmt.Println("node counter:", s.nodectr)
+	fmt.Println("alive nodes:", len(s.nodes))
+
+	// for _, n := range s.nodes {
+	// 	fmt.Printf("%x\n", n.tab.self.ID[:8])
+	// 	transport := n.conn.(*simTransport)
+	// 	fmt.Println("   joined:", transport.joinTime)
+	// 	fmt.Println("   sends:", transport.hashctr)
+	// 	fmt.Println("   table size:", n.tab.count)
+	// }
+}
+
+func (s *simulation) launchNode() *Network {
+	var (
+		num = s.nodectr
+		key = newkey()
+		id  = PubkeyID(&key.PublicKey)
+		ip  = make(net.IP, 4)
+	)
+	s.nodectr++
+	binary.BigEndian.PutUint32(ip, num)
+	ip[0] = 10
+	addr := &net.UDPAddr{IP: ip, Port: 30303}
+
+	transport := &simTransport{joinTime: time.Now(), sender: id, senderAddr: addr, sim: s}
+	net, err := newNetwork(transport, key.PublicKey, nil, "<no database>")
+	if err != nil {
+		panic("cannot launch new node: %v")
+	}
+	s.mu.Lock()
+	s.nodes[id] = net
+	s.mu.Unlock()
+
+	return net
+}
+
+func (s *simulation) dropNode(id NodeID) {
+	s.mu.Lock()
+	n := s.nodes[id]
+	delete(s.nodes, id)
+	s.mu.Unlock()
+
+	n.Close()
+}
+
+type simTransport struct {
+	joinTime   time.Time
+	sender     NodeID
+	senderAddr *net.UDPAddr
+	sim        *simulation
+	hashctr    uint64
+}
+
+func (st *simTransport) localAddr() *net.UDPAddr {
+	return st.senderAddr
+}
+
+func (st *simTransport) Close() {}
+
+func (st *simTransport) sendPing(remote *Node, remoteAddr *net.UDPAddr) []byte {
+	hash := st.nextHash()
+	st.sendPacket(remote.ID, ingressPacket{
+		remoteID:   st.sender,
+		remoteAddr: st.senderAddr,
+		hash:       hash,
+		ev:         pingPacket,
+		data: &ping{
+			Version:    4,
+			From:       rpcEndpoint{IP: st.senderAddr.IP, UDP: uint16(st.senderAddr.Port), TCP: 30303},
+			To:         rpcEndpoint{IP: remoteAddr.IP, UDP: uint16(remoteAddr.Port), TCP: 30303},
+			Expiration: uint64(time.Now().Unix() + int64(expiration)),
+		},
+	})
+	return hash
+}
+
+func (st *simTransport) sendPong(remote *Node, pingHash []byte) {
+	raddr := remote.addr()
+	st.sendPacket(remote.ID, ingressPacket{
+		remoteID:   st.sender,
+		remoteAddr: st.senderAddr,
+		hash:       st.nextHash(),
+		ev:         pongPacket,
+		data: &pong{
+			To:         rpcEndpoint{IP: raddr.IP, UDP: uint16(raddr.Port), TCP: 30303},
+			ReplyTok:   pingHash,
+			Expiration: uint64(time.Now().Unix() + int64(expiration)),
+		},
+	})
+}
+
+func (st *simTransport) sendFindnode(remote *Node, target NodeID) {
+	st.sendPacket(remote.ID, ingressPacket{
+		remoteID:   st.sender,
+		remoteAddr: st.senderAddr,
+		hash:       st.nextHash(),
+		ev:         findnodePacket,
+		data: &findnode{
+			Target:     target,
+			Expiration: uint64(time.Now().Unix() + int64(expiration)),
+		},
+	})
+}
+
+func (st *simTransport) sendNeighbours(remote *Node, nodes []*Node) {
+	// TODO: send multiple packets
+	rnodes := make([]rpcNode, len(nodes))
+	for i := range nodes {
+		rnodes[i] = nodeToRPC(nodes[i])
+	}
+	st.sendPacket(remote.ID, ingressPacket{
+		remoteID:   st.sender,
+		remoteAddr: st.senderAddr,
+		hash:       st.nextHash(),
+		ev:         neighborsPacket,
+		data: &neighbors{
+			Nodes:      rnodes,
+			Expiration: uint64(time.Now().Unix() + int64(expiration)),
+		},
+	})
+}
+
+func (st *simTransport) nextHash() []byte {
+	v := atomic.AddUint64(&st.hashctr, 1)
+	var hash common.Hash
+	binary.BigEndian.PutUint64(hash[:], v)
+	return hash[:]
+}
+
+func (st *simTransport) sendPacket(remote NodeID, p ingressPacket) {
+	st.sim.mu.Lock()
+	recipient := st.sim.nodes[remote]
+	st.sim.mu.Unlock()
+
+	// TODO: apply packet loss
+	time.AfterFunc(100*time.Millisecond, func() {
+		recipient.reqReadPacket(p)
+	})
+}

p2p/discover/sim_testmain_test.go

@@ -0,0 +1,43 @@
+// Copyright 2016 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+// +build go1.4,nacl,faketime_simulation
+
+package discover
+
+import (
+	"os"
+	"runtime"
+	"testing"
+	"unsafe"
+)
+
+// Enable fake time mode in the runtime, like on the go playground.
+// There is a slight chance that this won't work because some go code
+// might have executed before the variable is set.
+
+//go:linkname faketime runtime.faketime
+var faketime = 1
+
+func TestMain(m *testing.M) {
+	// We need to use unsafe somehow in order to get access to go:linkname.
+	_ = unsafe.Sizeof(0)
+
+	// Run the actual test. runWithPlaygroundTime ensures that the only test
+	// that runs is the one calling it.
+	runtime.GOMAXPROCS(8)
+	os.Exit(m.Run())
+}