Merge 23973c994d into 881bdc289f

p2p/protocols/protocol.go

@@ -0,0 +1,320 @@
+/*
+Package protocols is an extension to p2p. It offers a user friendly simple way to define
+devp2p subprotocols by abstracting away code standardly shared by protocols.
+The subprotocol architecture is inspired by the node package. Similar to a node
+the standard protocol (class) registers service contructors that are instantiated as service
+instances on the protocol isntance that is launched on a p2p peer connection.
+
+By mounting various protocol modules protocols can encapsulate vertical slices of business logic
+without duplicating code related to protocol communication.
+Standard protocol supports:
+
+* mounting services instantiated with the remote peer when a protocol instance is launched on a newly
+  established peer connection
+* registering module-specific handshakes and offers validation and renegotiation  of handshakes
+* registering multiple handlers for incoming messages
+* automate assigments of code indexes to messages
+* automate RLP decoding/encoding based on reflecting
+* provide the forever loop to read incoming messages
+* standardise error handling related to communication
+* enables access to sister services of the same peer connection analogous to node.Service
+* TODO: automatic generation of wire protocol specification for peers
+* peerPool abstracting out peer management by defining a peerPool that is called to register/unregister
+  peers as they connect and drop (ideally the peerPool also implements the peerPool interface that the
+  p2p server needs to suggest peers to connect to in server-as-initiator mode of operation
+  see https://github.com/ethereum/go-ethereum/issues/2254 for the peer management/connectivity related
+  aspect
+
+*/
+
+package protocols
+
+import (
+	"fmt"
+	"reflect"
+
+	"github.com/ethereum/go-ethereum/logger"
+	"github.com/ethereum/go-ethereum/logger/glog"
+	"github.com/ethereum/go-ethereum/p2p"
+)
+
+// error codes used by this  protocol scheme
+const (
+	ErrMsgTooLong = iota
+	ErrDecode
+	ErrWrite
+	ErrInvalidMsgCode
+	ErrInvalidMsgType
+	ErrLocalHandshake
+	ErrRemoteHandshake
+	ErrHandler
+)
+
+// error description strings associated with the codes
+var errorToString = map[int]string{
+	ErrMsgTooLong:      "Message too long",
+	ErrDecode:          "Invalid message (RLP error)",
+	ErrWrite:           "Error sending message",
+	ErrInvalidMsgCode:  "Invalid message code",
+	ErrInvalidMsgType:  "Invalid message type",
+	ErrLocalHandshake:  "Local handshake error",
+	ErrRemoteHandshake: "Remote handshake error",
+	ErrHandler:         "Message handler error",
+}
+
+/*
+Error implements the standard go error interface.
+Use:
+
+  errorf(code, format, params ...interface{})
+
+Prints as:
+
+ <description>: <details>
+
+where description is given by code in errorToString
+and details is fmt.Sprintf(format, params...)
+
+exported field Code can be checked
+*/
+type Error struct {
+	Code    int
+	message string
+	format  string
+	params  []interface{}
+}
+
+func (self Error) Error() (message string) {
+	if len(message) == 0 {
+		name, ok := errorToString[self.Code]
+		if !ok {
+			panic("invalid message code")
+		}
+		self.message = name
+		if self.format != "" {
+			self.message += ": " + fmt.Sprintf(self.format, self.params...)
+		}
+	}
+	return self.message
+}
+
+func errorf(code int, format string, params ...interface{}) *Error {
+	self := &Error{
+		Code:   code,
+		format: format,
+		params: params,
+	}
+
+	return self
+}
+
+// implements the code table spec
+// listing the message codes and types etc
+// and further metadata about the protocol
+type CodeMap struct {
+	Name       string                // name of the protocol
+	Version    uint                  // version
+	MaxMsgSize int                   // max length of message payload size
+	codes      []reflect.Type        // index of codes to msg types - to create zero values
+	messages   map[reflect.Type]uint // index of types to codes, for sending by type
+}
+
+func NewCodeMap(name string, version uint, maxMsgSize int, msgs ...interface{}) *CodeMap {
+	self := &CodeMap{
+		Name:       name,
+		Version:    version,
+		MaxMsgSize: maxMsgSize,
+		messages:   make(map[reflect.Type]uint),
+	}
+	self.Register(msgs...)
+	return self
+}
+
+func (self *CodeMap) Length() uint64 {
+	return uint64(len(self.codes))
+}
+
+func (self *CodeMap) Register(msgs ...interface{}) {
+	code := uint(len(self.codes))
+	for _, msg := range msgs {
+		typ := reflect.TypeOf(msg)
+		_, found := self.messages[typ]
+		if found {
+			// ignore duplicates
+			continue
+		}
+		// next code assigned to message type typ
+		self.messages[typ] = code
+		self.codes = append(self.codes, typ)
+		code++
+	}
+}
+
+// A Peer represents a remote peer or protocol instance that is running on a peer connection with
+// a remote peer
+type Peer struct {
+	ct         *CodeMap                                   // CodeMap for the protocol
+	*p2p.Peer                                             // the p2p.Peer object representing the remote
+	rw         p2p.MsgReadWriter                          // p2p.MsgReadWriter to send messages to and read messages from
+	handlers   map[reflect.Type][]func(interface{}) error //  message type -> message handler callback(s) map
+	disconnect func()                                     // Disconnect function set differently for testing
+	// lastActive time.Time                                  // tracking last active state
+}
+
+// NewPeer returns a new peer
+// this constructor is called by the p2p.Protocol#Run function
+// the first two arguments are comming the arguments passed to p2p.Protocol.Run function
+// the third argument is the CodeMap describing the protocol messages and options
+func NewPeer(p *p2p.Peer, rw p2p.MsgReadWriter, ct *CodeMap) *Peer {
+	return newPeer(p, rw, ct, func() {
+		p.Disconnect(p2p.DiscSubprotocolError)
+	})
+}
+
+// the disconnect function needs to be set differently for testing
+func NewTestPeer(p *p2p.Peer, rw p2p.MsgReadWriter, ct *CodeMap) *Peer {
+	return newPeer(p, rw, ct, func() {
+		rw.(*p2p.MsgPipeRW).Close()
+	})
+}
+
+func newPeer(p *p2p.Peer, rw p2p.MsgReadWriter, ct *CodeMap, disconn func()) *Peer {
+	return &Peer{
+		ct:         ct,
+		Peer:       p,
+		rw:         rw,
+		handlers:   make(map[reflect.Type][]func(interface{}) error),
+		disconnect: disconn,
+	}
+}
+
+// Register is called on the peer typically within the constructor of service instances running on peer connections
+// These constructors are called by the p2p.Protocol#Run function
+// It ties handler callbackss for specific message types
+// A message type can have several handlers registered by the same or different protocol services
+// Register is meant to be called once, deregistering is not currently supported therefore
+// handlers are assumed to be static across handshake renegotiations
+// i.e., a service instance either handles a message or not (irrespective of the handshake)
+// it panics if the message type is not defined in the CodeMap
+func (self *Peer) Register(msg interface{}, handler func(interface{}) error) uint {
+	typ := reflect.TypeOf(msg)
+	code, found := self.ct.messages[typ]
+	if !found {
+		panic(fmt.Sprintf("message type '%v' unknown ", typ))
+	}
+	glog.V(logger.Debug).Infof("registered handle for %v %v", msg, typ)
+	self.handlers[typ] = append(self.handlers[typ], handler)
+	return code
+}
+
+// Run starts the forever loop that handles incoming messages
+// called within the p2p.Protocol#Run function
+func (self *Peer) Run() error {
+	var err error
+	for {
+		_, err = self.handleIncoming()
+		if err != nil {
+			return err
+		}
+	}
+}
+
+// Drop disconnects a peer.
+// falls back to self.disconnect which is set as p2p.Peer#Disconnect except
+// for test peers where it calls p2p.MsgPipe#Close so that the readloop can terminate
+// TODO: may need to implement protocol drop only? don't want to kick off the peer
+// if they are useful for other protocols
+// overwrite Disconnect for testing, so that protocol readloop quits
+func (self *Peer) Drop() {
+	self.disconnect()
+}
+
+// Send takes a message, encodes it in RLP, finds the right message code and sends the
+// message off to the peer
+// this low level call will be wrapped by libraries providing routed or broadcast sends
+// but often just used to forward and push messages to directly connected peers
+func (self *Peer) Send(msg interface{}) error {
+	typ := reflect.TypeOf(msg)
+	code, found := self.ct.messages[typ]
+	if !found {
+		return errorf(ErrInvalidMsgType, "%v", typ)
+	}
+	glog.V(logger.Debug).Infof("=> %v %v (%d)", msg, typ, code)
+	err := p2p.Send(self.rw, uint64(code), msg)
+	if err != nil {
+		self.Drop()
+		return errorf(ErrWrite, "(msg code: %v): %v", code, err)
+	}
+	return nil
+}
+
+// handleIncoming(code)
+// is called each cycle of the main forever loop that handles and dispatches incoming messages
+// if this returns an error the loop returns and the peer is disconnected with the error
+// checks message size, out-of-range message codes, handles decoding with reflection,
+// call handlers as callback onside
+func (self *Peer) handleIncoming() (interface{}, error) {
+	msg, err := self.rw.ReadMsg()
+	glog.V(logger.Debug).Infof("<= %v", msg)
+	if err != nil {
+		return nil, err
+	}
+	// make sure that the payload has been fully consumed
+	defer msg.Discard()
+
+	if msg.Size > uint32(self.ct.MaxMsgSize) {
+		return nil, errorf(ErrMsgTooLong, "%v > %v", msg.Size, self.ct.MaxMsgSize)
+	}
+
+	// check if the message code is correct
+	maxMsgCode := uint(len(self.ct.messages))
+	if msg.Code >= uint64(maxMsgCode) {
+		return nil, errorf(ErrInvalidMsgCode, "%v (>=%v)", msg.Code, maxMsgCode)
+	}
+
+	// it is safe to be unsafe here
+	typ := self.ct.codes[msg.Code]
+	val := reflect.New(typ)
+	req := val.Elem()
+	req.Set(reflect.Zero(typ))
+	if err := msg.Decode(val.Interface()); err != nil {
+		return nil, errorf(ErrDecode, "<= %v: %v", msg, err)
+	}
+	glog.V(logger.Debug).Infof("<= %v %v (%d)", req, typ, msg.Code)
+
+	// call the registered handler callbacks
+	// a registered callback take the decoded message as argument as an interface
+	// which the handler is supposed to cast to the appropriate type
+	// it is entirely safe not to check the cast in the handler since the handler is
+	// chosen based on the proper type in the first place
+	for _, f := range self.handlers[typ] {
+		glog.V(6).Infof("handler for %v", typ)
+		err = f(req.Interface())
+		if err != nil {
+			return nil, errorf(ErrHandler, "(msg code %v): %v", msg.Code, err)
+		}
+	}
+	return req.Interface(), nil
+}
+
+// Handshake initiates a handshake on the peer connection
+// * the argument is the local  handshake 	to be sent to the remote peer
+// * expects a remote handshake back of the same type
+// returns the remote hs and an error
+func (self *Peer) Handshake(hs interface{}) (interface{}, error) {
+	typ := reflect.TypeOf(hs)
+	_, found := self.ct.messages[typ]
+	if !found {
+		return nil, errorf(ErrLocalHandshake, "unknown handshake message type: %v", typ)
+	}
+	err := self.Send(hs)
+	if err != nil {
+		return nil, errorf(ErrLocalHandshake, "cannot send: %v", err)
+	}
+	// receiving and validating remote handshake, expect code
+	rhs, err := self.handleIncoming()
+	if err != nil {
+		return nil, errorf(ErrRemoteHandshake, "'%v': %v", self.ct.Name, err)
+	}
+	return rhs, nil
+}

p2p/protocols/protocol_test.go

@@ -0,0 +1,372 @@
+package protocols
+
+import (
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/p2p"
+	"github.com/ethereum/go-ethereum/p2p/discover"
+	p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
+)
+
+// handshake message type
+type hs0 struct {
+	C uint
+}
+
+// message to kill/drop the peer with index C
+type kill struct {
+	C discover.NodeID
+}
+
+// message to drop connection
+type drop struct {
+}
+
+// example peerPool to demonstrate registration of peer connections
+type peerPool struct {
+	lock  sync.Mutex
+	peers map[discover.NodeID]*Peer
+}
+
+func newPeerPool() *peerPool {
+	return &peerPool{peers: make(map[discover.NodeID]*Peer)}
+}
+
+func (self *peerPool) add(p *Peer) {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	self.peers[p.ID()] = p
+}
+
+func (self *peerPool) remove(p *Peer) {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	delete(self.peers, p.ID())
+}
+
+func (self *peerPool) has(n discover.NodeID) bool {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	_, ok := self.peers[n]
+	return ok
+}
+
+func (self *peerPool) get(n discover.NodeID) *Peer {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	return self.peers[n]
+}
+
+/// protoHandshake represents module-independent aspects of the protocol and is
+// the first message peers send and receive as part the initial exchange
+type protoHandshake struct {
+	Version   uint   // local and remote peer should have identical version
+	NetworkId string // local and remote  peer should have identical network id
+}
+
+// function to check local and remote  protoHandshake matches
+func checkProtoHandshake(local, remote *protoHandshake) error {
+
+	if remote.NetworkId != local.NetworkId {
+		return fmt.Errorf("%s (!= %s)", remote.NetworkId, local.NetworkId)
+	}
+
+	if remote.Version != local.Version {
+		return fmt.Errorf("%d (!= %d)", remote.Version, local.Version)
+	}
+	return nil
+}
+
+const networkId = "420"
+
+// newProtocol sets up a protocol
+// the run function here demonstrates a typical protocol using peerPool, handshake
+// and messages registered to handlers
+func newProtocol() (*p2p.Protocol, *peerPool) {
+	ct := NewCodeMap("test", 42, 1024, &protoHandshake{}, &hs0{}, &kill{}, &drop{})
+	pp := newPeerPool()
+
+	run := func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
+		peer := NewTestPeer(p, rw, ct)
+
+		// demonstrates use of peerPool, killing another peer connection as a response to a message
+		peer.Register(&kill{}, func(msg interface{}) error {
+			// panics if target.C out of range
+			id := msg.(*kill).C
+			// name := fmt.Sprintf("test-%d", i)
+			pp.get(id).Drop()
+			return nil
+		})
+
+		// for testing we can trigger self induced disconnect upon receiving drop message
+		peer.Register(&drop{}, func(msg interface{}) error {
+			return fmt.Errorf("received disconnect request")
+		})
+
+		// initiate one-off protohandshake and check validity
+		phs := &protoHandshake{ct.Version, networkId}
+		hs, err := peer.Handshake(phs)
+		if err != nil {
+			return err
+		}
+		rhs := hs.(*protoHandshake)
+		err = checkProtoHandshake(phs, rhs)
+		if err != nil {
+			return err
+		}
+
+		lhs := &hs0{1}
+		// module handshake demonstrating a simple repeatable exchange of same-type message
+		hs, err = peer.Handshake(lhs)
+		if err != nil {
+			return err
+		}
+		rmhs := hs.(*hs0)
+		if rmhs.C != lhs.C {
+			return fmt.Errorf("handshake mismatch remote %v != local %v", rmhs.C, lhs.C)
+		}
+
+		peer.Register(lhs, func(msg interface{}) error {
+			rhs := msg.(*hs0)
+			if rhs.C != lhs.C {
+				return fmt.Errorf("handshake mismatch remote %v != local %v", rhs.C, lhs.C)
+			}
+			return peer.Send(rhs)
+		})
+
+		// add/remove peer from pool
+		pp.add(peer)
+		defer pp.remove(peer)
+		// this launches a forever read loop
+		return peer.Run()
+	}
+
+	return &p2p.Protocol{
+		Name:    ct.Name,
+		Length:  uint64(len(ct.codes)),
+		Version: 42,
+		Run:     run,
+	}, pp
+}
+
+func protoHandshakeExchange(proto *protoHandshake) []p2ptest.Exchange {
+
+	return []p2ptest.Exchange{
+		p2ptest.Exchange{
+			Expects: []p2ptest.Expect{
+				p2ptest.Expect{
+					Code: 0,
+					Msg:  &protoHandshake{42, "420"},
+					Peer: 0,
+				},
+			},
+		},
+		p2ptest.Exchange{
+			Triggers: []p2ptest.Trigger{
+				p2ptest.Trigger{
+					Code: 0,
+					Msg:  proto,
+					Peer: 0,
+				},
+			},
+		},
+	}
+}
+
+func runProtoHandshake(t *testing.T, proto *protoHandshake, err error) {
+	p, _ := newProtocol()
+	ids := p2ptest.RandomNodeIDs(t, 1)
+	s := p2ptest.NewSession(t, p, ids, nil)
+
+	s.TestExchanges(protoHandshakeExchange(proto)...)
+	s.TestDisconnects(err)
+}
+
+func TestProtoHandshakeVersionMismatch(t *testing.T) {
+	runProtoHandshake(t, &protoHandshake{41, "420"}, fmt.Errorf("41 (!= 42)"))
+}
+
+func TestProtoHandshakeNetworkIdMismatch(t *testing.T) {
+	runProtoHandshake(t, &protoHandshake{42, "421"}, fmt.Errorf("421 (!= 420)"))
+}
+
+func TestProtoHandshakeSuccess(t *testing.T) {
+	runProtoHandshake(t, &protoHandshake{42, "420"}, nil)
+}
+
+func moduleHandshakeExchange(resp uint) []p2ptest.Exchange {
+
+	return []p2ptest.Exchange{
+		p2ptest.Exchange{
+			Expects: []p2ptest.Expect{
+				p2ptest.Expect{
+					Code: 1,
+					Msg:  &hs0{1},
+					Peer: 0,
+				},
+			},
+		},
+		p2ptest.Exchange{
+			Triggers: []p2ptest.Trigger{
+				p2ptest.Trigger{
+					Code: 1,
+					Msg:  &hs0{resp},
+					Peer: 0,
+				},
+			},
+		},
+	}
+}
+
+func runModuleHandshake(t *testing.T, resp uint, err error) {
+	p, _ := newProtocol()
+	ids := p2ptest.RandomNodeIDs(t, 1)
+	s := p2ptest.NewSession(t, p, ids, nil)
+
+	s.TestExchanges(protoHandshakeExchange(&protoHandshake{42, "420"})...)
+	s.TestExchanges(moduleHandshakeExchange(resp)...)
+	s.TestDisconnects(err)
+}
+
+func TestModuleHandshakeError(t *testing.T) {
+	runModuleHandshake(t, 42, fmt.Errorf("handshake mismatch remote 42 != local 1"))
+}
+
+func TestModuleHandshakeSuccess(t *testing.T) {
+	runModuleHandshake(t, 1, nil)
+}
+
+// testing complex interactions over multiple peers, relaying, dropping
+func testMultiPeerSetup() []p2ptest.Exchange {
+
+	return []p2ptest.Exchange{
+		p2ptest.Exchange{
+			Expects: []p2ptest.Expect{
+				p2ptest.Expect{
+					Code: 0,
+					Msg:  &protoHandshake{42, "420"},
+					Peer: 0,
+				},
+				p2ptest.Expect{
+					Code: 0,
+					Msg:  &protoHandshake{42, "420"},
+					Peer: 1,
+				},
+			},
+		},
+		p2ptest.Exchange{
+			Triggers: []p2ptest.Trigger{
+				p2ptest.Trigger{
+					Code: 0,
+					Msg:  &protoHandshake{42, "420"},
+					Peer: 0,
+				},
+				p2ptest.Trigger{
+					Code: 0,
+					Msg:  &protoHandshake{42, "420"},
+					Peer: 1,
+				},
+			},
+			Expects: []p2ptest.Expect{
+				p2ptest.Expect{
+					Code: 1,
+					Msg:  &hs0{1},
+					Peer: 0,
+				},
+				p2ptest.Expect{
+					Code: 1,
+					Msg:  &hs0{1},
+					Peer: 1,
+				},
+			},
+		},
+		p2ptest.Exchange{
+			Triggers: []p2ptest.Trigger{
+				p2ptest.Trigger{
+					Code: 1,
+					Msg:  &hs0{1},
+					Peer: 0,
+				},
+				p2ptest.Trigger{
+					Code: 1,
+					Msg:  &hs0{1},
+					Peer: 1,
+				},
+			},
+		},
+	}
+}
+
+func runMultiplePeers(t *testing.T, ids []discover.NodeID, peer int, errs ...error) {
+	p, pp := newProtocol()
+	wg := &sync.WaitGroup{}
+	s := p2ptest.NewSession(t, p, ids, wg)
+
+	s.TestExchanges(testMultiPeerSetup()...)
+	// after some exchanges of messages, we can test state changes
+	// here this is simply demonstrated by the peerPool
+	// after the handshake negotiations peers must be addded to the pool
+	if !pp.has(ids[0]) {
+		t.Fatalf("missing peer test-0: %v", pp)
+	}
+	if !pp.has(ids[1]) {
+		t.Fatalf("missing peer test-1: %v", pp)
+	}
+
+	// sending kill request for peer with index <peer>
+	s.TestExchanges(p2ptest.Exchange{
+		Triggers: []p2ptest.Trigger{
+			p2ptest.Trigger{
+				Code: 2,
+				Msg:  &kill{ids[peer]},
+				Peer: 0,
+			},
+		},
+	})
+
+	// dropping the remaining peer
+	s.TestExchanges(p2ptest.Exchange{
+		Triggers: []p2ptest.Trigger{
+			p2ptest.Trigger{
+				Code: 3,
+				Msg:  &drop{},
+				Peer: (peer + 1) % 2,
+			},
+		},
+	})
+
+	// since drops are asyncronous, for correct testing you need to wait
+	//for all disconnections and error registration to complete or time out
+	errc := make(chan bool)
+	go func() {
+		wg.Wait()
+		close(errc)
+	}()
+
+	select {
+	case <-errc:
+	case <-time.NewTimer(1000 * time.Millisecond).C:
+		t.Fatalf("timed out")
+	}
+
+	// test if disconnected peers have been removed from peerPool
+	if pp.has(ids[peer]) {
+		t.Fatalf("peer test-% not dropped: %v", peer, pp)
+	}
+	// check the actual discconnect errors on the individual peers
+	s.TestDisconnects(errs...)
+
+}
+
+func TestMultiplePeersDropSelf(t *testing.T) {
+	ids := p2ptest.RandomNodeIDs(t, 2)
+	runMultiplePeers(t, ids, 0, fmt.Errorf("p2p: read or write on closed message pipe"), fmt.Errorf("Message handler error: (msg code 3): received disconnect request"))
+}
+
+func TestMultiplePeersDropOther(t *testing.T) {
+	ids := p2ptest.RandomNodeIDs(t, 2)
+	runMultiplePeers(t, ids, 1, fmt.Errorf("Message handler error: (msg code 3): received disconnect request"), fmt.Errorf("p2p: read or write on closed message pipe"))
+}

p2p/testing/protocols.go

@@ -0,0 +1,223 @@
+// Package protocols helpers_test make it easier to
+// write protocol tests by providing convenience functions and structures
+// protocols uses these helpers for  its own tests
+// but ideally should sit in p2p/protocols/testing/ subpackage
+package protocols
+
+import (
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/logger/glog"
+	"github.com/ethereum/go-ethereum/p2p"
+	"github.com/ethereum/go-ethereum/p2p/discover"
+)
+
+// a session represents a protocol running on multiple peer connections with single local node
+type Session struct {
+	IDs   []discover.NodeID
+	Peers []*p2p.MsgPipeRW
+	Errs  []error
+	t     *testing.T
+}
+
+// exchanges are the basic units of protocol tests
+// an exchange is defined on a session
+type Exchange struct {
+	Triggers []Trigger
+	Expects  []Expect
+}
+
+// part of the exchange, incoming message from a set of peers
+type Trigger struct {
+	Msg     interface{}   // type of message to be sent
+	Code    uint64        // code of message is given
+	Peer    int           // the peer to send the message to
+	Timeout time.Duration // timeout duration for the sending
+}
+
+type Expect struct {
+	Msg     interface{}   // type of message to expect
+	Code    uint64        // code of message is now given
+	Peer    int           // the peer-connection index to expect the message from
+	Timeout time.Duration // timeout duration of receiving
+}
+
+func randomNodeID(t *testing.T) (id discover.NodeID) {
+	key, err := crypto.GenerateKey()
+	if err != nil {
+		t.Fatalf("unable to generate key")
+	}
+	pubkey := crypto.FromECDSAPub(&key.PublicKey)
+	copy(id[:], pubkey)
+	return
+}
+
+func RandomNodeIDs(t *testing.T, n int) []discover.NodeID {
+	var ids []discover.NodeID
+	for i := 0; i < 2; i++ {
+		ids = append(ids, randomNodeID(t))
+	}
+	return ids
+}
+
+// NewSession creates a session by setting up a local peer with a prescribed set of peers
+// wg if present allows wg.Wait() be used to block until all peers disconnect
+// disconnect reason errors are written in session.Errs (correcponding to session,Peers)
+func NewSession(t *testing.T, protocol *p2p.Protocol, ids []discover.NodeID, wg *sync.WaitGroup) *Session {
+	peerCount := len(ids)
+	self := &Session{t: t}
+	caps := []p2p.Cap{p2p.Cap{protocol.Name, protocol.Version}}
+	if wg != nil {
+		wg.Add(peerCount)
+	}
+	run := func(j int, rws []p2p.MsgReadWriter) {
+		name := fmt.Sprintf("test-%d", j)
+		self.Errs[j] = protocol.Run(p2p.NewPeer(ids[j], name, caps), rws[j])
+		if wg != nil {
+			wg.Done()
+		}
+	}
+	var rws []p2p.MsgReadWriter
+	// connect peerCount number of peers
+	for i := 0; i < peerCount; i++ {
+		rw, rrw := p2p.MsgPipe()
+		self.Peers = append(self.Peers, rrw)
+		self.Errs = append(self.Errs, nil)
+		rws = append(rws, rw)
+	}
+	// start protocols on each peer connection
+	for i := 0; i < peerCount; i++ {
+		go run(i, rws)
+	}
+	return self
+}
+
+// trigger sends messages from peers
+func (self Session) trigger(trig Trigger) error {
+	if self.Errs[trig.Peer] != nil {
+		return fmt.Errorf("peer %v already disconnected with %v", trig.Peer, self.Errs[trig.Peer])
+	}
+	errc := make(chan error)
+	go func() {
+		errc <- p2p.Send(self.Peers[trig.Peer], trig.Code, trig.Msg)
+	}()
+
+	t := trig.Timeout
+	if t == time.Duration(0) {
+		t = 1000 * time.Millisecond
+	}
+	alarm := time.NewTimer(t)
+	select {
+	case err := <-errc:
+		return err
+	case <-alarm.C:
+		return fmt.Errorf("timout expecting %v to send to peer %v", trig.Msg, trig.Peer)
+	}
+}
+
+// expect checks an expectation
+func (self Session) expect(exp Expect) error {
+	if exp.Msg == nil {
+		panic("no message to expect")
+	}
+	if exp.Peer >= len(self.Errs) {
+		panic(fmt.Sprintf("peer %v does not exist: %v", exp.Peer))
+	}
+	if self.Errs[exp.Peer] != nil {
+		panic(fmt.Sprintf("peer %v already disconnected with: %v", exp.Peer, self.Errs))
+	}
+	errc := make(chan error)
+	go func() {
+		glog.V(6).Infof("waiting for msg, %v", exp.Msg)
+		errc <- p2p.ExpectMsg(self.Peers[exp.Peer], exp.Code, exp.Msg)
+	}()
+
+	t := exp.Timeout
+	if t == time.Duration(0) {
+		t = 1000 * time.Millisecond
+	}
+	alarm := time.NewTimer(t)
+	select {
+	case err := <-errc:
+		glog.V(6).Infof("expected msg arrives with error %v", err)
+		return err
+	case <-alarm.C:
+		glog.V(6).Infof("caught timeout")
+		return fmt.Errorf("timout expecting %v sent to peer %v", exp.Msg, exp.Peer)
+	}
+	// fatal upon encountering first exchange error
+}
+
+// TestExchange tests a series of exchanges againsts the session
+func (self Session) TestExchanges(exchanges ...Exchange) {
+	// launch all triggers of this exchanges
+
+	for i, e := range exchanges {
+		errc := make(chan error)
+		wg := &sync.WaitGroup{}
+		for _, trig := range e.Triggers {
+			wg.Add(1)
+			// separate go routing to allow parallel requests
+			go func(t Trigger) {
+				defer wg.Done()
+				err := self.trigger(t)
+				i++
+				if err != nil {
+					errc <- err
+				}
+			}(trig)
+		}
+
+		// each expectation is spawned in separate go-routine
+		// expectations of an exchange are conjunctive but uordered, i.e., only all of them arriving constitutes a pass
+		// each expectation is meant to be for a different peer, otherwise they are expected to panic
+		// testing of an exchange blocks until all expectations are decided
+		// an expectation is decided if
+		//  expected message arrives OR
+		// an unexpected message arrives (panic)
+		// times out on their individual tiemeout
+		for _, ex := range e.Expects {
+			wg.Add(1)
+			// expect msg spawned to separate go routine
+			go func(exp Expect) {
+				defer wg.Done()
+				err := self.expect(exp)
+				if err != nil {
+					glog.V(6).Infof("expect msg fails %v", err)
+					errc <- err
+				}
+			}(ex)
+		}
+
+		// wait for all expectations
+		go func() {
+			wg.Wait()
+			close(errc)
+		}()
+
+		// time out globally or finish when all expectations satisfied
+		alarm := time.NewTimer(500 * time.Millisecond)
+		select {
+
+		case err := <-errc:
+			glog.V(6).Infof("expectations finished with %v", err)
+			if err != nil {
+				self.t.Fatalf("exchange failed with: %v", err)
+			}
+		case <-alarm.C:
+			self.t.Fatalf("exchange timed out")
+		}
+	}
+}
+
+func (self Session) TestDisconnects(errs ...error) {
+	for i, err := range errs {
+		if !((err == nil && self.Errs[i] == nil) || err != nil && self.Errs[i] != nil && err.Error() == self.Errs[i].Error()) {
+			self.t.Fatalf("unexpected error on peer %v: '%v', wanted '%v'", i, self.Errs[i], err)
+		}
+	}
+}