diff --git a/p2p/adapters/msgpipes.go b/p2p/adapters/msgpipes.go new file mode 100644 index 0000000000..e9bff333a9 --- /dev/null +++ b/p2p/adapters/msgpipes.go @@ -0,0 +1,58 @@ +// 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 . + +package adapters + +import ( + "github.com/ethereum/go-ethereum/p2p" +) + +//network adapter's messenger interace +// NewPipe() (p2p.MsgReadWriter, p2p.MsgReadWriter) +// ClosePipe(rw p2p.MsgReadWriter) + +// protocol Messenger interface +// SendMsg(p2p.MsgWriter, uint64, interface{}) error +// ReadMsg(p2p.MsgReader) (p2p.Msg, error) + +// peer sesssion test +// ExpectMsg(p2p.MsgReader, uint64, interface{}) error +// SendMsg(p2p.MsgWriter, uint64, interface{}) error +type SimPipe struct{} + +func (*SimPipe) NewPipe() (p2p.MsgReadWriter, p2p.MsgReadWriter) { + return p2p.MsgPipe() +} + +func (*SimPipe) ClosePipe(rw p2p.MsgReadWriter) { + rw.(*p2p.MsgPipeRW).Close() +} + +func (*SimPipe) SendMsg(w p2p.MsgWriter, code uint64, msg interface{}) error { + return p2p.Send(w, code, msg) +} + +func (*SimPipe) ReadMsg(r p2p.MsgReader) (p2p.Msg, error) { + return r.ReadMsg() +} + +func (*SimPipe) TriggerMsg(w p2p.MsgWriter, code uint64, msg interface{}) error { + return p2p.Send(w, code, msg) +} + +func (*SimPipe) ExpectMsg(r p2p.MsgReader, code uint64, msg interface{}) error { + return p2p.ExpectMsg(r, code, msg) +} diff --git a/p2p/adapters/rlpx.go b/p2p/adapters/rlpx.go new file mode 100644 index 0000000000..8119cf226a --- /dev/null +++ b/p2p/adapters/rlpx.go @@ -0,0 +1,88 @@ +// 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 . + +package adapters + +import ( + "fmt" + "net" + + "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/p2p/discover" +) + +// devp2p RLPx underlay support + +type RLPx struct { + net *p2p.Server + addr []byte +} + +type RPLxMessenger struct{} + +func (*RPLxMessenger) SendMsg(w p2p.MsgWriter, code uint64, msg interface{}) error { + return p2p.Send(w, code, msg) +} + +func (*RPLxMessenger) ReadMsg(r p2p.MsgReader) (p2p.Msg, error) { + return r.ReadMsg() +} + +func (self *RLPx) LocalAddr() []byte { + return self.addr +} + +// func (self *RLPx) NewPeer(p *p2p.Peer, rw p2p.MsgReadWriter, ct *protocols.CodeMap) *protocols.Peer { +// return protocols.NewPeer(p, rw, ct, self, func() {}) +// } + +func (self *RLPx) Connect(enode []byte) error { + // TCP/UDP node address encoded with enode url scheme + // @:(?udp=) + node, err := discover.ParseNode(string(enode)) + if err != nil { + return fmt.Errorf("invalid node URL: %v", err) + } + self.net.AddPeer(node) + return nil +} + +func (self *RLPx) Disconnect(p *p2p.Peer, rw p2p.MsgReadWriter) { + p.Disconnect(p2p.DiscSubprotocolError) +} + +// ParseAddr take two arguments, advertised in handshake and the one set on the peer struct +// and constructs the remote address object +func (self *RLPx) ParseAddr(s []byte, remoteAddr string) ([]byte, error) { + + // returns self advertised node connection info (listening address w enodes) + // IP will get repaired on the other end if missing + // or resolved via ID by discovery at dialout + n, err := discover.ParseNode(string(s)) + if err != nil { + return nil, err + } + + // repair reported address if IP missing + if n.IP.IsUnspecified() { + host, _, err := net.SplitHostPort(remoteAddr) + if err != nil { + return nil, err + } + n.IP = net.ParseIP(host) + } + return []byte(n.String()), nil +} diff --git a/p2p/adapters/simnet.go b/p2p/adapters/simnet.go new file mode 100644 index 0000000000..4c200917d5 --- /dev/null +++ b/p2p/adapters/simnet.go @@ -0,0 +1,171 @@ +// 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 . + +package adapters + +import ( + "fmt" + "sync" + + "github.com/ethereum/go-ethereum/logger/glog" + "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/p2p/discover" +) + +type NetAdapter interface { + Connect([]byte) error + Disconnect(*p2p.Peer, p2p.MsgReadWriter) + LocalAddr() []byte + ParseAddr([]byte, string) ([]byte, error) +} + +func newPeer(rw p2p.MsgReadWriter) *Peer { + return &Peer{ + RW: rw, + Errc: make(chan error, 1), + Flushc: make(chan bool), + Onc: make(chan bool), + } +} + +type Peer struct { + RW p2p.MsgReadWriter + Errc chan error + Flushc chan bool + Onc chan bool +} + +// Network interface to retrieve protocol runner to launch upon peer +// connection +type Network interface { + Protocol(id *discover.NodeID) ProtoCall +} + +type Messenger interface { + SendMsg(p2p.MsgWriter, uint64, interface{}) error + ReadMsg(p2p.MsgReader) (p2p.Msg, error) + NewPipe() (p2p.MsgReadWriter, p2p.MsgReadWriter) + ClosePipe(rw p2p.MsgReadWriter) +} + +type ProtoCall func(*p2p.Peer, p2p.MsgReadWriter) error + +func NewSimNet(id *discover.NodeID, n Network, m Messenger) *SimNet { + return &SimNet{ + ID: id, + Network: n, + Messenger: m, + PeerMap: make(map[discover.NodeID]int), + } +} + +// Simnet is the network adapter that +type SimNet struct { + ID *discover.NodeID + Network + Messenger + Run ProtoCall + PeerMap map[discover.NodeID]int + Peers []*Peer + lock sync.RWMutex +} + +func Key(id []byte) string { + return string(id) +} + +func Name(id []byte) string { + return fmt.Sprintf("test-%08x", id) +} + +func (self *SimNet) LocalAddr() []byte { + return self.ID[:] +} + +func (self *SimNet) ParseAddr(p []byte, s string) ([]byte, error) { + return p, nil +} + +func (self *SimNet) GetPeer(id *discover.NodeID) *Peer { + self.lock.Lock() + defer self.lock.Unlock() + return self.getPeer(id) +} + +func (self *SimNet) getPeer(id *discover.NodeID) *Peer { + i, found := self.PeerMap[*id] + if !found { + return nil + } + return self.Peers[i] +} + +func (self *SimNet) SetPeer(id *discover.NodeID, rw p2p.MsgReadWriter) { + self.lock.Lock() + defer self.lock.Unlock() + self.setPeer(id, rw) +} + +func (self *SimNet) setPeer(id *discover.NodeID, rw p2p.MsgReadWriter) { + i, found := self.PeerMap[*id] + if !found { + i = len(self.Peers) + self.PeerMap[*id] = i + self.Peers = append(self.Peers, newPeer(rw)) + return + } + if self.Peers[i] != nil && rw != nil { + panic(fmt.Sprintf("pipe for %v already set", id)) + } + // legit reconnect reset disconnection error, + self.Peers[i].RW = rw +} + +func (self *SimNet) Disconnect(p *p2p.Peer, rw p2p.MsgReadWriter) { + self.lock.Lock() + defer self.lock.Unlock() + self.ClosePipe(rw) + id := p.ID() + self.getPeer(&id).RW = nil + glog.V(6).Infof("dropped peer %v", id) +} + +func (self *SimNet) Connect(rid []byte) error { + self.lock.Lock() + defer self.lock.Unlock() + var id discover.NodeID + copy(id[:], rid) + peer := self.getPeer(&id) + if peer != nil { + return fmt.Errorf("already connected") + } + run := self.Protocol(&id) + rw, rrw := self.NewPipe() + glog.V(6).Infof("connect to peer %v, setting pipe", id) + self.setPeer(&id, rrw) + if run != nil { + p := p2p.NewPeer(*self.ID, Name(self.ID[:]), []p2p.Cap{}) + go run(p, rrw) + } + peer = self.getPeer(&id) + go func() { + glog.V(6).Infof("simnet connect to %v", id) + p := p2p.NewPeer(id, Name(id[:]), []p2p.Cap{}) + err := self.Run(p, rw) + peer.Errc <- err + }() + return nil +} diff --git a/p2p/protocols/README.md b/p2p/protocols/README.md new file mode 100644 index 0000000000..7292acec39 --- /dev/null +++ b/p2p/protocols/README.md @@ -0,0 +1,27 @@ +p2p/protocols: devp2p subprotocol abstraction + +The protocols subpackage is an extension to p2p. It offers a simple and user friendly simple way +to define devp2p subprotocols by abstracting away code that implementations would typically share. + +The package provides a protocol peer object of type protocols.Peer initialised from + +* a p2p.Peer, a p2p.MsgReadWriter (the arguments passed to p2p.Protocol#Run), +* a protocols.CodeMap, this encodes the msg code and msg type associations +* messenger interface (with methods SendMsg and ReadMsg) that abstracts out sending and receiving a msg +* disconnect function + +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. + +Features + +* registering multiple handler callbacks for incoming messages +* automate RLP decoding/encoding based on reflection +* provide the forever loop to read incoming messages +* standardise error handling related to communication +* with disconnection and messaging abstracted out allows protocols to be used + in network simulations with or without serialisation, transport and p2p server +* TODO: automatic generation of wire protocol specification for peers + +see the possibly obsolete #2254 for the peer management/connectivity related aspect) diff --git a/p2p/protocols/protocol.go b/p2p/protocols/protocol.go new file mode 100644 index 0000000000..5d1e81a5c5 --- /dev/null +++ b/p2p/protocols/protocol.go @@ -0,0 +1,324 @@ +/* +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 + ErrNoHandler + 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", + ErrNoHandler: "No handler registered 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 + m Messenger // defines senf and receive + *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 +} + +type Messenger interface { + SendMsg(p2p.MsgWriter, uint64, interface{}) error + ReadMsg(p2p.MsgReader) (p2p.Msg, error) +} + +// 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, m Messenger, disconn func()) *Peer { + return &Peer{ + ct: ct, + m: m, + 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() + glog.V(6).Infof("handled it: %v", err) + 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 := self.m.SendMsg(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) { + glog.V(6).Infof("handle incoming..") + msg, err := self.m.ReadMsg(self.rw) + glog.V(6).Infof("got err: %v", err) + if err != nil { + return nil, err + } + glog.V(logger.Debug).Infof("<= %v", msg) + // 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 + handlers := self.handlers[typ] + if len(handlers) == 0 { + glog.V(6).Infof("no handler (msg code %v)", msg.Code) + // return nil, errorf(ErrNoHandler, "(msg code %v)", msg.Code) + } else { + for i, f := range handlers { + glog.V(6).Infof("handler %v for %v", i, 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..9c6fbcb4b5 --- /dev/null +++ b/p2p/protocols/protocol_test.go @@ -0,0 +1,361 @@ +package protocols + +import ( + "fmt" + "sync" + "testing" + + "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/p2p/adapters" + "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 nodeID C +type kill struct { + C *discover.NodeID +} + +// message to drop connection +type drop struct { +} + +/// 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 +} + +// checkProtoHandshake verifies local and remote protoHandshakes match +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(pp *p2ptest.TestPeerPool, wg *sync.WaitGroup) func(adapters.NetAdapter, adapters.Messenger) adapters.ProtoCall { + ct := NewCodeMap("test", 42, 1024, &protoHandshake{}, &hs0{}, &kill{}, &drop{}) + + return func(na adapters.NetAdapter, m adapters.Messenger) adapters.ProtoCall { + return func(p *p2p.Peer, rw p2p.MsgReadWriter) error { + if wg != nil { + wg.Add(1) + } + peer := NewPeer(p, rw, ct, m, func() { na.Disconnect(p, rw) }) + + // demonstrates use of peerPool, killing another peer connection as a response to a message + peer.Register(&kill{}, func(msg interface{}) error { + id := msg.(*kill).C + 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{42} + // module handshake demonstrating a simple repeatable exchange of same-type message + hs, err = peer.Handshake(lhs) + if err != nil { + return err + } + + if rmhs := hs.(*hs0); 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) + } + lhs.C += rhs.C + return peer.Send(lhs) + }) + + // add/remove peer from pool + pp.Add(peer) + defer pp.Remove(peer) + // this launches a forever read loop + err = peer.Run() + if wg != nil { + wg.Done() + } + return err + } + } +} + +func protocolTester(t *testing.T, pp *p2ptest.TestPeerPool, wg *sync.WaitGroup) *p2ptest.ExchangeSession { + id := p2ptest.RandomNodeID() + return p2ptest.NewProtocolTester(t, id, 2, newProtocol(pp, wg)) +} + +func protoHandshakeExchange(id *discover.NodeID, proto *protoHandshake) []p2ptest.Exchange { + + return []p2ptest.Exchange{ + p2ptest.Exchange{ + Expects: []p2ptest.Expect{ + p2ptest.Expect{ + Code: 0, + Msg: &protoHandshake{42, "420"}, + Peer: id, + }, + }, + }, + p2ptest.Exchange{ + Triggers: []p2ptest.Trigger{ + p2ptest.Trigger{ + Code: 0, + Msg: proto, + Peer: id, + }, + }, + }, + } +} + +func runProtoHandshake(t *testing.T, proto *protoHandshake, errs ...error) { + pp := p2ptest.NewTestPeerPool() + s := protocolTester(t, pp, nil) + // TODO: make this more than one handshake + id := s.IDs[0] + s.TestExchanges(protoHandshakeExchange(id, proto)...) + var disconnects []*p2ptest.Disconnect + for i, err := range errs { + disconnects = append(disconnects, &p2ptest.Disconnect{Peer: s.IDs[i], Error: err}) + } + s.TestDisconnected(disconnects...) +} + +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"}) +} + +func moduleHandshakeExchange(id *discover.NodeID, resp uint) []p2ptest.Exchange { + + return []p2ptest.Exchange{ + p2ptest.Exchange{ + Expects: []p2ptest.Expect{ + p2ptest.Expect{ + Code: 1, + Msg: &hs0{42}, + Peer: id, + }, + }, + }, + p2ptest.Exchange{ + Triggers: []p2ptest.Trigger{ + p2ptest.Trigger{ + Code: 1, + Msg: &hs0{resp}, + Peer: id, + }, + }, + }, + } +} + +func runModuleHandshake(t *testing.T, resp uint, errs ...error) { + pp := p2ptest.NewTestPeerPool() + s := protocolTester(t, pp, nil) + id := s.IDs[0] + s.TestExchanges(protoHandshakeExchange(id, &protoHandshake{42, "420"})...) + s.TestExchanges(moduleHandshakeExchange(id, resp)...) + var disconnects []*p2ptest.Disconnect + for i, err := range errs { + disconnects = append(disconnects, &p2ptest.Disconnect{Peer: s.IDs[i], Error: err}) + } + s.TestDisconnected(disconnects...) +} + +func TestModuleHandshakeError(t *testing.T) { + runModuleHandshake(t, 43, fmt.Errorf("handshake mismatch remote 43 > local 42")) +} + +func TestModuleHandshakeSuccess(t *testing.T) { + runModuleHandshake(t, 42) +} + +// testing complex interactions over multiple peers, relaying, dropping +func testMultiPeerSetup(a, b *discover.NodeID) []p2ptest.Exchange { + + return []p2ptest.Exchange{ + p2ptest.Exchange{ + Expects: []p2ptest.Expect{ + p2ptest.Expect{ + Code: 0, + Msg: &protoHandshake{42, "420"}, + Peer: a, + }, + p2ptest.Expect{ + Code: 0, + Msg: &protoHandshake{42, "420"}, + Peer: b, + }, + }, + }, + p2ptest.Exchange{ + Triggers: []p2ptest.Trigger{ + p2ptest.Trigger{ + Code: 0, + Msg: &protoHandshake{42, "420"}, + Peer: a, + }, + p2ptest.Trigger{ + Code: 0, + Msg: &protoHandshake{42, "420"}, + Peer: b, + }, + }, + Expects: []p2ptest.Expect{ + p2ptest.Expect{ + Code: 1, + Msg: &hs0{42}, + Peer: a, + }, + p2ptest.Expect{ + Code: 1, + Msg: &hs0{42}, + Peer: b, + }, + }, + }, + p2ptest.Exchange{ + Triggers: []p2ptest.Trigger{ + p2ptest.Trigger{ + Code: 1, + Msg: &hs0{41}, + Peer: a, + }, + p2ptest.Trigger{ + Code: 1, + Msg: &hs0{41}, + Peer: b, + }, + }, + }, + p2ptest.Exchange{ + Triggers: []p2ptest.Trigger{ + p2ptest.Trigger{ + Code: 1, + Msg: &hs0{1}, + Peer: a, + }, + }, + }, + p2ptest.Exchange{ + Expects: []p2ptest.Expect{ + p2ptest.Expect{ + Code: 1, + Msg: &hs0{43}, + Peer: a, + }, + }, + }, + } +} + +func runMultiplePeers(t *testing.T, peer int, errs ...error) { + wg := &sync.WaitGroup{} + pp := p2ptest.NewTestPeerPool() + s := protocolTester(t, pp, wg) + + s.TestExchanges(testMultiPeerSetup(s.IDs[0], s.IDs[1])...) + // 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(s.IDs[0]) { + t.Fatalf("missing peer test-0: %v (%v)", pp, s.IDs) + } + if !pp.Has(s.IDs[1]) { + t.Fatalf("missing peer test-1: %v (%v)", pp, s.IDs) + } + + // sending kill request for peer with index + s.TestExchanges(p2ptest.Exchange{ + Triggers: []p2ptest.Trigger{ + p2ptest.Trigger{ + Code: 2, + Msg: &kill{s.IDs[peer]}, + Peer: s.IDs[0], + }, + }, + }) + + // dropping the remaining peer + s.TestExchanges(p2ptest.Exchange{ + Triggers: []p2ptest.Trigger{ + p2ptest.Trigger{ + Code: 3, + Msg: &drop{}, + Peer: s.IDs[(peer+1)%2], + }, + }, + }) + wg.Wait() + // check the actual discconnect errors on the individual peers + var disconnects []*p2ptest.Disconnect + for i, err := range errs { + disconnects = append(disconnects, &p2ptest.Disconnect{Peer: s.IDs[i], Error: err}) + } + s.TestDisconnected(disconnects...) + // test if disconnected peers have been removed from peerPool + if pp.Has(s.IDs[peer]) { + t.Fatalf("peer test-%v not dropped: %v (%v)", peer, pp, s.IDs) + } + +} + +func TestMultiplePeersDropSelf(t *testing.T) { + runMultiplePeers(t, 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) { + runMultiplePeers(t, 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/simulations/examples/connectivity.go b/p2p/simulations/examples/connectivity.go new file mode 100644 index 0000000000..9307637a02 --- /dev/null +++ b/p2p/simulations/examples/connectivity.go @@ -0,0 +1,22 @@ +package main + +import ( + "runtime" + + "github.com/ethereum/go-ethereum/logger/glog" + "github.com/ethereum/go-ethereum/p2p/simulations" +) + +// var server +func main() { + runtime.GOMAXPROCS(runtime.NumCPU()) + glog.SetV(6) + glog.SetToStderr(true) + + c, quitc := simulations.NewSessionController() + + simulations.StartRestApiServer("8888", c) + // wait until server shuts down + <-quitc + +} diff --git a/p2p/simulations/network.go b/p2p/simulations/network.go new file mode 100644 index 0000000000..3836117c0c --- /dev/null +++ b/p2p/simulations/network.go @@ -0,0 +1,391 @@ +package simulations + +import ( + "bytes" + "fmt" + "math/rand" + "sync" + "time" + + "github.com/ethereum/go-ethereum/logger/glog" + "github.com/ethereum/go-ethereum/p2p/adapters" + "github.com/ethereum/go-ethereum/p2p/discover" +) + +const lablen = 4 + +func NewNetworkController(n *Network, parent *ResourceController) Controller { + self := NewResourceContoller( + &ResourceHandlers{ + // Destroy: n.Shutdown, nil + // Create: n.StartNode, NodeConfig + // Update: n.Setup, NodeConfig + // Retrieve: n.Retrieve, + Retrieve: &ResourceHandler{ + Handle: n.Query, + }, + }, + ) + if parent != nil { + parent.SetResource(fmt.Sprintf("%d", parent.id), self) + } + // self.SetResource("nodes", NewNodesController()) + return Controller(self) +} + +// Network +// this can be the hook for uptime +type Network struct { + adapters.Messenger + lock sync.RWMutex + NodeMap map[discover.NodeID]int + Nodes []*SimNode + Journal []*Entry +} + +func NewNetwork(m adapters.Messenger) *Network { + return &Network{ + Messenger: m, + NodeMap: make(map[discover.NodeID]int), + } +} + +type SimNode struct { + ID *discover.NodeID + config *NodeConfig + NetAdapter adapters.NetAdapter +} + +func (self *SimNode) String() string { + return fmt.Sprintf("SimNode %v", self.ID.String()[0:lablen]) +} + +func (self *SimConn) String() string { + return fmt.Sprintf("SimConn %v->%v", self.Caller.String()[0:lablen], self.Callee.String()[0:lablen]) +} + +type NodeConfig struct { + ID *discover.NodeID + Run func(adapters.NetAdapter, adapters.Messenger) adapters.ProtoCall +} + +func Key(id []byte) string { + return string(id) +} + +func (self *Network) Protocol(id *discover.NodeID) adapters.ProtoCall { + self.lock.Lock() + defer self.lock.Unlock() + node := self.getNode(id) + if node == nil { + return nil + } + na := node.NetAdapter.(*adapters.SimNet) + return na.Run +} + +// TODO: ignored for now +type QueryConfig struct { + Format string // "cy.update", "journal", +} + +type CyData struct { + Id string `json:"id"` + Source string `json:"source"` + Target string `json:"target"` + On bool `json:"on"` +} + +type CyElement struct { + Data *CyData `json:"data"` + Classes string `json:"classes"` + Group string `json:"group"` + // selected: false, // whether the element is selected (default false) + // selectable: true, // whether the selection state is mutable (default true) + // locked: false, // when locked a node's position is immutable (default false) + // grabbable: true, // whether the node can be grabbed and moved by the user +} + +type Entry struct { + Action string `json:"action"` + Type string `json:"type"` + Object interface{} `json:"object"` +} + +func (self *Entry) Stirng() string { + return fmt.Sprintf("\n", self.Action, self.Type, self.Object) +} + +func (n *Network) AppendEntries(entries ...*Entry) { + n.lock.Lock() + n.Journal = append(n.Journal, entries...) + n.lock.Unlock() +} + +type Know struct { + Subject *discover.NodeID `json:"subject"` + Object *discover.NodeID `json:"objectr"` + // Into + // number of attempted connections + // time of attempted connections + // number of active connections during the session + // number of active connections since records began + // swap balance +} + +// active connections are represented by the SimNode entry object so that +// you journal updates could filter if passive knowledge about peers is +// irrelevant +type SimConn struct { + Caller *discover.NodeID `json:"caller"` + Callee *discover.NodeID `json:"callee"` + // Info + // active connection + // average throughput, recent average throughput +} + +func (self *Network) CyUpdate() *CyUpdate { + self.lock.Lock() + defer self.lock.Unlock() + added := []*CyElement{} + removed := []string{} + var el *CyElement + for _, entry := range self.Journal { + glog.V(6).Infof("journal entry: %v", entry) + switch entry.Type { + case "Node": + el = &CyElement{Group: "nodes", Data: &CyData{Id: entry.Object.(*SimNode).ID.String()[0:lablen]}} + case "Conn": + // mutually exclusive directed edge (caller -> callee) + source := entry.Object.(*SimConn).Caller.String()[0:lablen] + target := entry.Object.(*SimConn).Callee.String()[0:lablen] + first := source + second := target + if bytes.Compare([]byte(first), []byte(second)) > 1 { + first = target + second = source + } + id := fmt.Sprintf("%v-%v", first, second) + el = &CyElement{Group: "edges", Data: &CyData{Id: id, Source: source, Target: target}} + case "Know": + // independent directed edge (peer0 registers peer1) + source := entry.Object.(*Know).Subject.String()[0:lablen] + target := entry.Object.(*Know).Object.String()[0:lablen] + id := fmt.Sprintf("%v-%v-%v", source, target, "know") + el = &CyElement{Group: "edges", Data: &CyData{Id: id, Source: source, Target: target}} + } + switch entry.Action { + case "Add": + added = append(added, el) + case "Remove": + removed = append(removed, el.Data.Id) + case "On": + el.Data.On = true + added = append(added, el) + case "Off": + el.Data.On = false + removed = append(removed, el.Data.Id) + } + } + self.Journal = nil + return &CyUpdate{ + Add: added, + Remove: removed, + } +} + +type CyUpdate struct { + Add []*CyElement `json:"add"` + Remove []string `json:"remove"` +} + +func (self *Network) Query(conf interface{}, c *ResourceController) (interface{}, error) { + glog.V(6).Infof("query: GET handler ") + // config := conf.(*QueryConfig) + return interface{}(self.CyUpdate()), nil +} + +// deltas: changes in the number of cumulative actions: non-negative integers. +// base unit is the fixed minimal interval between two measurements (time quantum) +// acceleration : to slow down you just set the base unit higher. +// to speed up: skip x number of base units +// frequency: given as the (constant or average) number of base units between measurements +// if resolution is expressed as the inverse of frequency = preserved information +// setting the acceleration +// beginning of the record (lifespan) of the network is index 0 +// acceleration means that snapshots are rarer so the same span can be generated by the journal +// then update logs can be compressed (toonly one state transition per affected node) +// epoch, epochcount + +type Delta struct { + On int + Off int +} + +func oneOutOf(n int) int { + t := rand.Intn(n) + if t == 0 { + return 1 + } + return 0 +} + +func deltas(i int) (d []*Delta) { + if i == 0 { + return []*Delta{ + &Delta{10, 0}, + &Delta{20, 0}, + } + } + return []*Delta{ + &Delta{oneOutOf(10), oneOutOf(10)}, + &Delta{oneOutOf(2), oneOutOf(2)}, + } +} + +func mockJournalTest(nw *Network, ticker *<-chan time.Time) { + + ids := RandomNodeIDs(100) + action := "Off" + for n := 0; ; n++ { + select { + case <-*ticker: + var entries []*Entry + + if n == 0 { + + entries = []*Entry{ + &Entry{ + Type: "Node", + Action: "On", + Object: &SimNode{ID: ids[0]}, + }, + &Entry{ + Type: "Node", + Action: "On", + Object: &SimNode{ID: ids[1]}, + }, + } + } else { + sc := &SimConn{ + Caller: ids[0], + Callee: ids[1], + } + if n%3 == 0 { + if action == "On" { + action = "Off" + } else { + action = "On" + } + entries = append(entries, &Entry{ + Type: "Conn", + Action: action, + Object: sc, + }) + } + } + + glog.V(6).Info("entries: %v", entries) + nw.AppendEntries(entries...) + } + } +} + +func mockJournal(nw *Network, ticker *<-chan time.Time) { + ids := RandomNodeIDs(100) + var onNodes []*SimNode + offNodes := ids + var onConns []*SimConn + + for n := 0; ; n++ { + select { + case <-*ticker: + var entries []*Entry + ds := deltas(n) + for i := 0; len(offNodes) > 0 && i < ds[0].On; i++ { + c := rand.Intn(len(offNodes)) + sn := &SimNode{ID: offNodes[c]} + entries = append(entries, &Entry{ + Type: "Node", + Action: "On", + Object: sn, + }) + onNodes = append(onNodes, sn) + offNodes = append(offNodes[0:c], offNodes[c+1:]...) + } + for i := 0; len(onNodes) > 0 && i < ds[0].Off; i++ { + c := rand.Intn(len(onNodes)) + sn := onNodes[c] + entries = append(entries, &Entry{ + Type: "Node", + Action: "Off", + Object: sn, + }) + onNodes = append(onNodes[0:c], onNodes[c+1:]...) + offNodes = append(offNodes, sn.ID) + } + for i := 0; len(onNodes) > 1 && i < ds[1].On; i++ { + caller := onNodes[rand.Intn(len(onNodes))].ID + callee := onNodes[rand.Intn(len(onNodes))].ID + if caller == callee { + i-- + continue + } + sc := &SimConn{ + Caller: caller, + Callee: callee, + } + entries = append(entries, &Entry{ + Type: "Conn", + Action: "On", + Object: sc, + }) + onConns = append(onConns, sc) + } + for i := 0; len(onConns) > 0 && i < ds[1].Off; i++ { + c := rand.Intn(len(onConns)) + entries = append(entries, &Entry{ + Type: "Conn", + Action: "Off", + Object: onConns[c], + }) + onConns = append(onConns[0:c], onConns[c+1:]...) + } + glog.V(6).Info("entries: %v", entries) + nw.AppendEntries(entries...) + } + } + +} + +func (self *Network) StartNode(conf *NodeConfig) error { + self.lock.Lock() + defer self.lock.Unlock() + id := conf.ID + + _, found := self.NodeMap[*id] + if found { + return fmt.Errorf("node %v already running", id) + } + simnet := adapters.NewSimNet(id, self, self) + if conf.Run != nil { + simnet.Run = conf.Run(simnet, self.Messenger) + } + self.NodeMap[*id] = len(self.Nodes) + self.Nodes = append(self.Nodes, &SimNode{id, conf, adapters.NetAdapter(simnet)}) + return nil +} + +func (self *Network) GetNode(id *discover.NodeID) *SimNode { + self.lock.Lock() + defer self.lock.Unlock() + return self.getNode(id) +} + +func (self *Network) getNode(id *discover.NodeID) *SimNode { + i, found := self.NodeMap[*id] + if !found { + return nil + } + return self.Nodes[i] +} diff --git a/p2p/simulations/rest_api_server.go b/p2p/simulations/rest_api_server.go new file mode 100644 index 0000000000..8b7d9e9225 --- /dev/null +++ b/p2p/simulations/rest_api_server.go @@ -0,0 +1,68 @@ +package simulations + +import ( + "fmt" + "io/ioutil" + "net/http" + "strings" + "time" + + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/logger/glog" +) + +type Controller interface { + Resource(id string) (Controller, error) + Handle(method string) (returnHandler, error) + SetResource(id string, c Controller) +} + +// starts up http server +func StartRestApiServer(port string, c Controller) { + serveMux := http.NewServeMux() + serveMux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { + handle(w, r, c) + }) + go http.ListenAndServe(":"+port, serveMux) + glog.V(logger.Info).Infof("Swarm Network Controller HTTP server started on localhost:%s", port) +} + +func handle(w http.ResponseWriter, r *http.Request, c Controller) { + requestURL := r.URL + glog.V(logger.Debug).Infof("HTTP %s request URL: '%s', Host: '%s', Path: '%s', Referer: '%s', Accept: '%s'", r.Method, r.RequestURI, requestURL.Host, requestURL.Path, r.Referer(), r.Header.Get("Accept")) + uri := requestURL.Path + w.Header().Set("Content-Type", "text/json") + w.Header().Set("Access-Control-Allow-Origin", "*") + defer r.Body.Close() + parts := strings.Split(uri, "/") + var err error + for _, id := range parts { + if len(id) == 0 { + continue + } + glog.V(6).Infof("server: resolving to controller for resource id '%v'", id) + c, err = c.Resource(id) + if err != nil { + http.Error(w, fmt.Sprintf("resource %v not found", id), http.StatusNotFound) + return + } + } + handler, err := c.Handle(r.Method) + if err != nil { + http.Error(w, fmt.Sprintf("method %v not allowed (%v)", r.Method, err), http.StatusMethodNotAllowed) + return + } + glog.V(6).Infof("server: calling controller handler on body") + // on return we close the request Body so we assume it is read synchronously + response, err := handler(r.Body) + var resp []byte + if response != nil { + resp, err = ioutil.ReadAll(response) + } + glog.V(6).Infof("server: called controller handler on body, response: %v", string(resp)) + if err != nil { + http.Error(w, fmt.Sprintf("handler error: %v", err), http.StatusBadRequest) + return + } + http.ServeContent(w, r, "", time.Now(), response) +} diff --git a/p2p/simulations/rest_api_server_test.go b/p2p/simulations/rest_api_server_test.go new file mode 100644 index 0000000000..9d74c531ef --- /dev/null +++ b/p2p/simulations/rest_api_server_test.go @@ -0,0 +1,137 @@ +package simulations + +import ( + "bytes" + "fmt" + "io" + "io/ioutil" + "net/http" + "testing" +) + +const testPort = "8889" + +type testController struct { +} + +func (self *testController) SetResource(id string, c Controller) { +} + +func (self *testController) Resource(id string) (Controller, error) { + if id == "missing" { + return nil, fmt.Errorf("missing") + } + return Controller(self), nil +} + +func (self *testController) Handle(method string) (returnHandler, error) { + switch method { + case "POST": + case "DELETE": + default: + return nil, fmt.Errorf("allowed methods: POST DELETE") + } + return handlerf(method), nil +} + +func handlerf(method string) returnHandler { + return func(r io.Reader) (io.ReadSeeker, error) { + body, err := ioutil.ReadAll(r) + if err != nil { + return nil, err + } + if string(body) == "invalid" { + return nil, fmt.Errorf("invalid body") + } + return io.ReadSeeker(bytes.NewReader([]byte("response"))), nil + } +} + +func init() { + StartRestApiServer(testPort, &testController{}) +} + +type testRequest struct { + method string + path string + body string + response string + status int +} + +type ReadCloser struct { + io.Reader +} + +func (ReadCloser) Close() {} + +func testResponses(t *testing.T, reqs ...*testRequest) { + for _, req := range reqs { + path := url(testPort, req.path) + var r *http.Response + var err error + switch req.method { + case "POST": + r, err = http.Post(path, "text/json", ReadCloser{bytes.NewReader([]byte(req.body))}) + default: + r, err = http.Get(path) + } + if err != nil { + t.Fatalf("unexpected error on request: %v", err) + } + if r.StatusCode != req.status { + t.Fatalf("unexpected status on request: got %v, expected %v", r.StatusCode, req.status) + } + body, err := ioutil.ReadAll(r.Body) + if err != nil { + t.Fatalf("unexpected error on reading body: %v", err) + } + if string(body) != req.response { + t.Fatalf("unexpected response body. got '%s', expected '%v'", body, req.response) + } + } +} + +func TestServerMethodNotAllowed(t *testing.T) { + testResponses(t, + &testRequest{ + "GET", + "anypath", + "anybody", + "method GET not allowed (allowed methods: POST DELETE)\n", + http.StatusMethodNotAllowed, + }) +} + +func TestServerInvalid(t *testing.T) { + testResponses(t, + &testRequest{ + "POST", + "anypath", + "invalid", + "handler error: invalid body\n", + http.StatusBadRequest, + }) +} + +func TestServerResourceNotFound(t *testing.T) { + testResponses(t, + &testRequest{ + "POST", + "missing", + "anybody", + "resource missing not found\n", + http.StatusNotFound, + }) +} + +func TestServerSuccess(t *testing.T) { + testResponses(t, + &testRequest{ + "POST", + "anypath", + "anybody", + "response", + http.StatusOK, + }) +} diff --git a/p2p/simulations/session_controller.go b/p2p/simulations/session_controller.go new file mode 100644 index 0000000000..754110744c --- /dev/null +++ b/p2p/simulations/session_controller.go @@ -0,0 +1,169 @@ +package simulations + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "io/ioutil" + "reflect" + "sync" + "time" + + "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/logger/glog" + "github.com/ethereum/go-ethereum/p2p/adapters" + "github.com/ethereum/go-ethereum/p2p/discover" +) + +type returnHandler func(io.Reader) (io.ReadSeeker, error) + +type ResourceHandler struct { + Handle func(interface{}, *ResourceController) (interface{}, error) + Type reflect.Type +} + +type ResourceHandlers struct { + Create, Retrieve, Update, Destroy *ResourceHandler +} + +type ResourceController struct { + lock sync.Mutex + controllers map[string]Controller + id int + methods []string + *ResourceHandlers +} + +var methodsAvailable = []string{"POST", "GET", "PUT", "DELETE"} + +func (self *ResourceHandlers) handler(method string) *ResourceHandler { + var h *ResourceHandler + switch method { + case "POST": + h = self.Create + case "GET": + h = self.Retrieve + case "PUT": + h = self.Update + case "DELETE": + h = self.Destroy + } + return h +} + +func NewResourceContoller(c *ResourceHandlers) *ResourceController { + var methods []string + for _, method := range methodsAvailable { + if c.handler(method) != nil { + methods = append(methods, method) + } + } + return &ResourceController{ + ResourceHandlers: c, + controllers: make(map[string]Controller), + methods: methods, + } +} + +func NewSessionController() (*ResourceController, chan bool) { + quitc := make(chan bool) + return NewResourceContoller( + &ResourceHandlers{ + + Create: &ResourceHandler{ + Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) { + // TODO: take config for type of network + network := NewNetwork(&adapters.SimPipe{}) + ticker := time.NewTicker(1000 * time.Millisecond) + go mockJournal(network, &ticker.C) + return NewNetworkController(network, parent), nil + }, + }, + + Destroy: &ResourceHandler{ + Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) { + glog.V(6).Infof("destroy handler called") + // this can quit the entire app (shut down the backend server) + quitc <- true + return nil, nil + }, + }, + }, + ), quitc +} + +func (self *ResourceController) Handle(method string) (returnHandler, error) { + h := self.handler(method) + if h == nil { + return nil, fmt.Errorf("allowed methods: %v", self.methods) + } + glog.V(6).Infof("get handler callback for method %v", method) + rh := func(r io.Reader) (io.ReadSeeker, error) { + input, err := ioutil.ReadAll(r) + if err != nil { + glog.V(6).Infof("reading json body: %v", err) + return nil, err + } + glog.V(6).Infof("decode json request body") + var arg interface{} + if h.Type != nil { + val := reflect.New(h.Type) + req := val.Elem() + req.Set(reflect.Zero(h.Type)) + err = json.Unmarshal(input, val.Interface()) + if err != nil { + return nil, err + } + arg = req.Interface() + } + glog.V(6).Infof("calling handler") + res, err := h.Handle(arg, self) + if err != nil { + return nil, err + } + resp, err := json.MarshalIndent(res, "", " ") + return bytes.NewReader(resp), nil + } + return rh, nil +} + +func (self *ResourceController) Resource(id string) (Controller, error) { + self.lock.Lock() + defer self.lock.Unlock() + c, ok := self.controllers[id] + glog.V(6).Infof("resource for id %v", id) + if !ok { + return nil, fmt.Errorf("not found") + } + return c, nil +} + +func (self *ResourceController) SetResource(id string, c Controller) { + self.lock.Lock() + defer self.lock.Unlock() + if c == nil { + delete(self.controllers, id) + } else { + self.controllers[id] = c + } +} + +func RandomNodeID() *discover.NodeID { + key, err := crypto.GenerateKey() + if err != nil { + panic("unable to generate key") + } + var id discover.NodeID + pubkey := crypto.FromECDSAPub(&key.PublicKey) + copy(id[:], pubkey[1:]) + return &id +} + +func RandomNodeIDs(n int) []*discover.NodeID { + var ids []*discover.NodeID + for i := 0; i < n; i++ { + ids = append(ids, RandomNodeID()) + } + return ids +} diff --git a/p2p/simulations/session_controller_test.go b/p2p/simulations/session_controller_test.go new file mode 100644 index 0000000000..7cc16570ea --- /dev/null +++ b/p2p/simulations/session_controller_test.go @@ -0,0 +1,121 @@ +package simulations + +import ( + "fmt" + "io/ioutil" + "net/http" + "testing" + "time" + + "github.com/ethereum/go-ethereum/logger/glog" + "github.com/ethereum/go-ethereum/p2p/discover" +) + +const ( + domain = "http://localhost" + port = "8888" +) + +var quitc chan bool +var controller *ResourceController + +func init() { + glog.SetV(6) + glog.SetToStderr(true) + controller, quitc = NewSessionController() + StartRestApiServer(port, controller) +} + +func url(port, path string) string { + return fmt.Sprintf("%v:%v/%v", domain, port, path) +} + +func TestQuit(t *testing.T) { + req, err := http.NewRequest("DELETE", url(port, ""), nil) + // req, err := http.NewRequest("PUT", url(""), nil) + if err != nil { + t.Fatalf("unexpected error") + } + var resp *http.Response + go func() { + r, err := (&http.Client{}).Do(req) + if err != nil { + t.Fatalf("unexpected error") + } + resp = r + }() + timeout := time.NewTimer(1000 * time.Millisecond) + select { + case <-quitc: + case <-timeout.C: + t.Fatalf("timed out: controller did not quit, response: %v", resp) + } +} + +func TestUpdate(t *testing.T) { + req, err := http.NewRequest("GET", url(port, "0"), nil) + if err != nil { + t.Fatalf("unexpected error") + } + keys := []string{ + "aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80", + "f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3", + } + var ids []*discover.NodeID + for _, key := range keys { + id := discover.MustHexID(key) + ids = append(ids, &id) + } + network := NewNetwork(nil) + NewNetworkController(network, controller) + Update(network, ids) + r, err := (&http.Client{}).Do(req) + if err != nil { + t.Fatalf("unexpected error") + } + resp, err := ioutil.ReadAll(r.Body) + if err != nil { + t.Fatalf("error reading response body: %v", err) + } + exp := `{ + "add": [ + { + "data": { + "id": "aa7c", + "source": "", + "target": "", + "on": false + }, + "classes": "", + "group": "nodes" + }, + { + "data": { + "id": "f5ae", + "source": "", + "target": "", + "on": false + }, + "classes": "", + "group": "nodes" + } + ], + "remove": [] +}` + if string(resp) != exp { + t.Fatalf("incorrect response body. got\n'%v', expected\n'%v'", string(resp), exp) + } +} + +func Update(self *Network, ids []*discover.NodeID) { + self.lock.Lock() + defer self.lock.Unlock() + for _, id := range ids { + e := &Entry{ + Action: "Add", + Type: "Node", + Object: &SimNode{ID: id, config: &NodeConfig{ID: id}}, + } + self.Journal = append(self.Journal, e) + } +} diff --git a/p2p/testing/exchange.go b/p2p/testing/exchange.go new file mode 100644 index 0000000000..659d77615a --- /dev/null +++ b/p2p/testing/exchange.go @@ -0,0 +1,304 @@ +// 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 testing + +import ( + "fmt" + "sync" + "testing" + "time" + + "github.com/ethereum/go-ethereum/logger/glog" + "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/p2p/adapters" + "github.com/ethereum/go-ethereum/p2p/discover" +) + +// ExchangeTestSession assumes a network with a protocol running on multiple peer connection +// and is used to test scanarios of message exchange among a select array of nodes +// the scenarios are sets of exchanges, each with a trigger and an expectation +// This rigid regime is suitable for +// * unit testing protocol message exchanges (nodes are peers of a local node) +// * testing routed messaging between remote non-connected nodes within a group +type ExchangeTestSession struct { + lock sync.Mutex + IDs []*discover.NodeID + TestNetAdapter + TestMessenger + t *testing.T +} + +// implemented by simulations/ +type TestNetAdapter interface { + GetPeer(id *discover.NodeID) *adapters.Peer +} + +type TestMessenger interface { + // MsgPipe([]byte, []byte) p2p,MsgPipe + ExpectMsg(p2p.MsgReader, uint64, interface{}) error + TriggerMsg(p2p.MsgWriter, uint64, interface{}) error +} + +// 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 *discover.NodeID // 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 *discover.NodeID // the peer that expects the message + Timeout time.Duration // timeout duration for receiving +} + +type Disconnect struct { + Peer *discover.NodeID // the peer that expects the message + Error error +} + +// NewExchangeTestSession takes a network session and Messenger +// and returns an exchange session test driver that can +// be used to unit test protocol communications +// it allows for resource-driven scenario testing +// disconnect reason errors are written in session.Errs +// (correcponding to session.Peers) +func NewExchangeTestSession(t *testing.T, n TestNetAdapter, m TestMessenger, ids []*discover.NodeID) *ExchangeTestSession { + return &ExchangeTestSession{ + IDs: ids, + TestNetAdapter: n, + TestMessenger: m, + t: t, + } +} + +type TestPeerInfo struct { + RW p2p.MsgReadWriter + Flushc chan bool + Errc chan error +} + +// trigger sends messages from peers +func (self *ExchangeTestSession) trigger(trig Trigger) error { + peer := self.GetPeer(trig.Peer) + if peer == nil { + panic(fmt.Sprintf("trigger: peer %v does not exist (1- %v)", trig.Peer, len(self.IDs))) + } + rw := peer.RW + if rw == nil { + return fmt.Errorf("trigger: peer %v unreachable", trig.Peer) + } + errc := make(chan error) + + go func() { + glog.V(6).Infof("trigger....") + errc <- self.TriggerMsg(rw, trig.Code, trig.Msg) + glog.V(6).Infof("triggered") + }() + + 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) + } +} + +func Key(id []byte) string { + return string(id) +} + +// expect checks an expectation +func (self *ExchangeTestSession) expect(exp Expect) error { + if exp.Msg == nil { + panic("no message to expect") + } + peer := self.GetPeer(exp.Peer) + if peer == nil { + panic(fmt.Sprintf("expect: peer %v does not exist (1- %v)", exp.Peer, len(self.IDs))) + } + rw := peer.RW + if rw == nil { + return fmt.Errorf("trigger: peer %v unreachable", exp.Peer) + } + + errc := make(chan error) + go func() { + glog.V(6).Infof("waiting for msg, %v", exp.Msg) + errc <- self.ExpectMsg(rw, 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 *ExchangeTestSession) 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) + 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(1000 * time.Millisecond) + select { + + case err := <-errc: + if err != nil { + self.t.Fatalf("exchange failed with: %v", err) + } else { + glog.V(6).Infof("exchange %v run successfully", i) + } + case <-alarm.C: + self.t.Fatalf("exchange timed out") + } + } +} + +type flushMsg struct{} + +func flushExchange(c int, ids ...*discover.NodeID) Exchange { + var triggers []Trigger + for _, id := range ids { + triggers = append(triggers, + Trigger{ + Code: uint64(c), + Msg: &flushMsg{}, + Peer: id, + }) + } + return Exchange{ + Triggers: triggers, + } +} + +var FlushMsg = &flushMsg{} + +func (self *ExchangeTestSession) TestConnected(flush bool, peers ...*discover.NodeID) { + timeout := time.NewTimer(1000 * time.Millisecond) + var flushc chan bool + if !flush { + flushc = make(chan bool) + close(flushc) + } + wg := &sync.WaitGroup{} + wg.Add(len(peers)) + for _, id := range peers { + glog.V(6).Infof("checking if peer %v is connected", id) + ticker := time.NewTicker(100 * time.Millisecond) + go func(p *discover.NodeID) { + defer wg.Done() + for { + peer := self.GetPeer(p) + if peer != nil { + if flush { + flushc = peer.Flushc + } + glog.V(6).Infof("checking if peer %v is connected", id) + select { + case <-timeout.C: + self.t.Fatalf("exchange timed out waiting for peer %v to flush", p) + case err := <-peer.Errc: + self.t.Fatalf("peer %v disconnected with error %v", p, err) + case <-flushc: + glog.V(6).Infof("peer %v is connected", p) + return + } + } + select { + case <-ticker.C: + glog.V(6).Infof("waiting for %v to connect", p) + case <-timeout.C: + self.t.Fatalf("exchange timed out waiting for peer %v to connect", p) + } + } + }(id) + } + wg.Wait() + glog.V(6).Infof("checking complete") + +} + +func (self *ExchangeTestSession) TestDisconnected(disconnects ...*Disconnect) { + for _, disconnect := range disconnects { + id := disconnect.Peer + err := disconnect.Error + errc := self.GetPeer(id).Errc + alarm := time.NewTimer(1000 * time.Millisecond) + select { + case derr := <-errc: + if !((err == nil && derr == nil) || err != nil && derr != nil && err.Error() == derr.Error()) { + self.t.Fatalf("unexpected error on peer %v: '%v', wanted '%v'", id, derr, err) + } + case <-alarm.C: + self.t.Fatalf("exchange timed out waiting for peer %v to disconnect", id) + } + } +} diff --git a/p2p/testing/peerpool.go b/p2p/testing/peerpool.go new file mode 100644 index 0000000000..52691e0f47 --- /dev/null +++ b/p2p/testing/peerpool.go @@ -0,0 +1,47 @@ +package testing + +import ( + "sync" + + "github.com/ethereum/go-ethereum/p2p/discover" +) + +type TestPeer interface { + ID() discover.NodeID + Drop() +} + +// TestPeerPool is an example peerPool to demonstrate registration of peer connections +type TestPeerPool struct { + lock sync.Mutex + peers map[discover.NodeID]TestPeer +} + +func NewTestPeerPool() *TestPeerPool { + return &TestPeerPool{peers: make(map[discover.NodeID]TestPeer)} +} + +func (self *TestPeerPool) Add(p TestPeer) { + self.lock.Lock() + defer self.lock.Unlock() + self.peers[p.ID()] = p +} + +func (self *TestPeerPool) Remove(p TestPeer) { + self.lock.Lock() + defer self.lock.Unlock() + delete(self.peers, p.ID()) +} + +func (self *TestPeerPool) Has(n *discover.NodeID) bool { + self.lock.Lock() + defer self.lock.Unlock() + _, ok := self.peers[*n] + return ok +} + +func (self *TestPeerPool) Get(n *discover.NodeID) TestPeer { + self.lock.Lock() + defer self.lock.Unlock() + return self.peers[*n] +} diff --git a/p2p/testing/sessions.go b/p2p/testing/sessions.go new file mode 100644 index 0000000000..3b5127ce75 --- /dev/null +++ b/p2p/testing/sessions.go @@ -0,0 +1,97 @@ +package testing + +import ( + "testing" + + "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/logger/glog" + "github.com/ethereum/go-ethereum/p2p/adapters" + "github.com/ethereum/go-ethereum/p2p/discover" + "github.com/ethereum/go-ethereum/p2p/simulations" +) + +type PeerAdapter interface { + adapters.NetAdapter + TestMessenger + TestNetAdapter +} + +type ExchangeSession struct { + network *simulations.Network + na adapters.NetAdapter + *ExchangeTestSession +} + +// NewProtocolTester returns an exchange test session +// this is a resource driver for protocol message exchange +// scenarios expressed as expects and triggers +// see p2p/protocols/exhange_test.go for an example +// this is used primarily to unit test protocols or protocol modules +// correct message exchange, forwarding, and broadcast +// higher level or network behaviour should be tested with network simulators +func NewProtocolTester(t *testing.T, id *discover.NodeID, n int, run func(adapters.NetAdapter, adapters.Messenger) adapters.ProtoCall) *ExchangeSession { + ids := RandomNodeIDs(n) + network := simulations.NewNetwork(&adapters.SimPipe{}) + + // setup a simulated network of n nodes + // Startup pivot node + err := network.StartNode(&simulations.NodeConfig{ID: id, Run: run}) + if err != nil { + panic(err.Error()) + } + na := network.GetNode(id).NetAdapter + s := NewExchangeTestSession(t, na.(TestNetAdapter), network.Messenger.(TestMessenger), nil) + self := &ExchangeSession{ + network: network, + na: na, + ExchangeTestSession: s, + } + self.Connect(ids...) + // Start up connections to virual nodes serving as endpoints for sending/receiving messages for peers + return self +} + +func (self *ExchangeTestSession) Flush(code int, ids ...*discover.NodeID) { + self.TestConnected(false, ids...) + glog.V(6).Infof("flushing peers %v (code %v)", ids, code) + self.TestExchanges(flushExchange(code, ids...)) + self.TestConnected(true, ids...) +} + +func (self *ExchangeSession) StartNode(id *discover.NodeID) error { + err := self.network.StartNode(&simulations.NodeConfig{ID: id, Run: nil}) + if err != nil { + return err + } + self.IDs = append(self.IDs, id) + return nil +} + +func (self *ExchangeSession) Connect(ids ...*discover.NodeID) { + for _, id := range ids { + glog.V(6).Infof("start node %v", id) + self.StartNode(id) + glog.V(6).Infof("connect to %v", id) + self.na.Connect(id[:]) + } + +} + +func RandomNodeID() *discover.NodeID { + key, err := crypto.GenerateKey() + if err != nil { + panic("unable to generate key") + } + var id discover.NodeID + pubkey := crypto.FromECDSAPub(&key.PublicKey) + copy(id[:], pubkey[1:]) + return &id +} + +func RandomNodeIDs(n int) []*discover.NodeID { + var ids []*discover.NodeID + for i := 0; i < n; i++ { + ids = append(ids, RandomNodeID()) + } + return ids +}