diff --git a/p2p/protocols/protocol.go b/p2p/protocols/protocol.go new file mode 100644 index 0000000000..20ca9cdc0b --- /dev/null +++ b/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: + + :
+ +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 +} diff --git a/p2p/protocols/protocol_test.go b/p2p/protocols/protocol_test.go new file mode 100644 index 0000000000..c152cd9a92 --- /dev/null +++ b/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 + 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")) +} diff --git a/p2p/testing/protocols.go b/p2p/testing/protocols.go index 7fc3fb72aa..9ba390fee1 100644 --- a/p2p/testing/protocols.go +++ b/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 { - self := &Session{t: t, Errs: make([]error, peerCount)} +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) } - 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 { - go f(i) + for i := 0; i < peerCount; 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() - ewg.Done() + go func() { + wg.Wait() + close(errc) + }() - } - // wait for all expectations - go func() { - ewg.Wait() - close(errc) - }() + // time out globally or finish when all expectations satisfied + alarm := time.NewTimer(500 * time.Millisecond) + select { - // time out globally or finish when all expectations satisfied - alarm := time.NewTimer(5000 * time.Millisecond) - select { - - case err := <-errc: - if err != nil { - self.t.Fatalf("exchange failed with: %v", err) + 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") } - case <-alarm.C: - self.t.Fatalf("exchange timed out") } }