p2p/protocols: devp2p subprotocol abstraction

Package protocols is an extension to p2p. It offers a user friendly simple way to define
devp2p subprotocols by abstracting away code that implementations would typically use.

The package provides a standard protocol peer object of type protocols.Peer initialised from
a p2p.Peer, a p2p.MsgReadWriter (the arguments passed to p2p.Protocol#Run
and a protocols.CodeMap.

Allowing the p2p.Protocol#Run function to construct this peer allows passing it to arbitrary
service instances sitting on peer connections. These service instances can encapsulate vertical slices
of business logic without duplicating code related to protocol communication.

*    registering multiple handler callbacks 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
*    TODO: automatic generation of wire protocol specification for peers

The test demonstrates the use of protocols/testing package for testing protocols

see the possibly obsolete #2254 for the peer management/connectivity related aspect)

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

@@ -11,12 +11,14 @@ import (
 	"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
@@ -44,16 +46,41 @@ type Expect struct {
 	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, peerCount int, wg *sync.WaitGroup) *Session {
+func NewSession(t *testing.T, protocol *p2p.Protocol, ids []discover.NodeID, wg *sync.WaitGroup) *Session {
-	self := &Session{t: t, Errs: make([]error, peerCount)}
+	peerCount := len(ids)
+	self := &Session{t: t}
 	caps := []p2p.Cap{p2p.Cap{protocol.Name, protocol.Version}}
 	if wg != nil {
 		wg.Add(peerCount)
 	}
-	var runs []func(int)
+	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++ {
@@ -61,24 +88,10 @@ func NewSession(t *testing.T, protocol *p2p.Protocol, peerCount int, wg *sync.Wa
 		self.Peers = append(self.Peers, rrw)
 		self.Errs = append(self.Errs, nil)
 		rws = append(rws, rw)
-		key, err := crypto.GenerateKey()
-		if err != nil {
-			t.Fatalf("unable to generate key")
-		}
-		pubkey := crypto.FromECDSAPub(&key.PublicKey)
-		var id discover.NodeID
-		copy(id[:], pubkey)
-		runs = append(runs, func(j int) {
-			name := fmt.Sprintf("test-%d", j)
-			self.Errs[j] = protocol.Run(p2p.NewPeer(id, name, caps), rws[j])
-			if wg != nil {
-				wg.Done()
-			}
-		})
 	}
 	// start protocols on each peer connection
-	for i, f := range runs {
+	for i := 0; i < peerCount; i++ {
-		go f(i)
+		go run(i, rws)
 	}
 	return self
 }
@@ -95,7 +108,7 @@ func (self Session) trigger(trig Trigger) error {
 
 	t := trig.Timeout
 	if t == time.Duration(0) {
-		t = 100 * time.Millisecond
+		t = 1000 * time.Millisecond
 	}
 	alarm := time.NewTimer(t)
 	select {
@@ -119,18 +132,21 @@ func (self Session) expect(exp Expect) error {
 	}
 	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 = 100 * time.Millisecond
+		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
@@ -140,10 +156,8 @@ func (self Session) expect(exp Expect) error {
 func (self Session) TestExchanges(exchanges ...Exchange) {
 	// launch all triggers of this exchanges
 
-	errc := make(chan error)
-	ewg := &sync.WaitGroup{}
 	for i, e := range exchanges {
-		ewg.Add(1)
+		errc := make(chan error)
 		wg := &sync.WaitGroup{}
 		for _, trig := range e.Triggers {
 			wg.Add(1)
@@ -173,32 +187,30 @@ func (self Session) TestExchanges(exchanges ...Exchange) {
 				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
-		wg.Wait()
+		go func() {
-		ewg.Done()
+			wg.Wait()
+			close(errc)
+		}()
 
-	}
+		// time out globally or finish when all expectations satisfied
-	// wait for all expectations
+		alarm := time.NewTimer(500 * time.Millisecond)
-	go func() {
+		select {
-		ewg.Wait()
-		close(errc)
-	}()
 
-	// time out globally or finish when all expectations satisfied
+		case err := <-errc:
-	alarm := time.NewTimer(5000 * time.Millisecond)
+			glog.V(6).Infof("expectations finished with %v", err)
-	select {
+			if err != nil {
-
+				self.t.Fatalf("exchange failed with: %v", err)
-	case err := <-errc:
+			}
-		if err != nil {
+		case <-alarm.C:
-			self.t.Fatalf("exchange failed with: %v", err)
+			self.t.Fatalf("exchange timed out")
 		}
-	case <-alarm.C:
-		self.t.Fatalf("exchange timed out")
 	}
 }