mirrors/tendermint
1
0
Fork 0
mirror of https://github.com/tendermint/tendermint.git synced 2026-09-20 06:54:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'develop' of https://github.com/tendermint/tendermint into develop

Browse Source 
Conflicts:
	p2p/peer.go
	rpc/core/net.go
	rpc/core/types/responses.go
This commit is contained in:
Ethan Buchman
2015-04-21 02:00:58 -07:00
parent 882a82bad4 aa931f88ab
commit 67ea49c5fb
29 changed files with 375 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files
p2p/connection.go
+17 -15
View File
@@ -11,7 +11,7 @@ import (
"time"
flow "code.google.com/p/mxk/go1/flowcontrol"
"github.com/tendermint/log15"
//"github.com/tendermint/log15"
"github.com/tendermint/tendermint/binary"
. "github.com/tendermint/tendermint/common"
)
@@ -184,7 +184,7 @@ func (c *MConnection) Send(chId byte, msg interface{}) bool {
return false
}
log.Debug("Send", "channel", chId, "connection", c, "msg", msg, "bytes", binary.BinaryBytes(msg))
log.Debug("Send", "channel", chId, "connection", c, "msg", msg) //, "bytes", binary.BinaryBytes(msg))
// Send message to channel.
channel, ok := c.channelsIdx[chId]
@@ -365,18 +365,20 @@ FOR_LOOP:
// Block until .recvMonitor says we can read.
c.recvMonitor.Limit(maxMsgPacketSize, atomic.LoadInt64(&c.recvRate), true)
// Peek into bufReader for debugging
if numBytes := c.bufReader.Buffered(); numBytes > 0 {
log.Debug("Peek connection buffer", "numBytes", numBytes, "bytes", log15.Lazy{func() []byte {
bytes, err := c.bufReader.Peek(MinInt(numBytes, 100))
if err == nil {
return bytes
} else {
log.Warn("Error peeking connection buffer", "error", err)
return nil
}
}})
}
/*
// Peek into bufReader for debugging
if numBytes := c.bufReader.Buffered(); numBytes > 0 {
log.Debug("Peek connection buffer", "numBytes", numBytes, "bytes", log15.Lazy{func() []byte {
bytes, err := c.bufReader.Peek(MinInt(numBytes, 100))
if err == nil {
return bytes
} else {
log.Warn("Error peeking connection buffer", "error", err)
return nil
}
}})
}
*/
// Read packet type
var n int64
@@ -417,7 +419,7 @@ FOR_LOOP:
}
msgBytes := channel.recvMsgPacket(pkt)
if msgBytes != nil {
log.Debug("Received bytes", "chId", pkt.ChannelId, "msgBytes", msgBytes)
//log.Debug("Received bytes", "chId", pkt.ChannelId, "msgBytes", msgBytes)
c.onReceive(pkt.ChannelId, msgBytes)
}
default:
p2p/listener.go
+4
View File
@@ -128,6 +128,10 @@ func (l *DefaultListener) ExternalAddress() *NetAddress {
return l.extAddr
}
func (l *DefaultListener) NetListener() net.Listener {
return l.listener
}
func (l *DefaultListener) Stop() {
if atomic.CompareAndSwapUint32(&l.stopped, 0, 1) {
l.listener.Close()
p2p/peer.go
+32 -3
View File
@@ -4,10 +4,12 @@ import (
"fmt"
"io"
"net"
"sync"
"sync/atomic"
"github.com/tendermint/tendermint/binary"
. "github.com/tendermint/tendermint/common"
"github.com/tendermint/tendermint/types"
)
type nodeInfo struct {
@@ -21,13 +23,39 @@ type Peer struct {
mconn *MConnection
running uint32
Nodeinfo *nodeInfo
*types.NodeInfo
Key string
Data *CMap // User data.
}
func newPeer(conn net.Conn, outbound bool, reactorsByCh map[byte]Reactor, chDescs []*ChannelDescriptor, onPeerError func(*Peer, interface{})) *Peer {
func peerHandshake(conn net.Conn, ourNodeInfo *types.NodeInfo) (*types.NodeInfo, error) {
var peerNodeInfo = new(types.NodeInfo)
var wg sync.WaitGroup
var err1 error
var err2 error
wg.Add(2)
go func() {
var n int64
binary.WriteBinary(ourNodeInfo, conn, &n, &err1)
wg.Done()
}()
go func() {
var n int64
binary.ReadBinary(peerNodeInfo, conn, &n, &err2)
log.Info("Peer handshake", "peerNodeInfo", peerNodeInfo)
wg.Done()
}()
wg.Wait()
if err1 != nil {
return nil, err1
}
if err2 != nil {
return nil, err2
}
return peerNodeInfo, nil
}
func newPeer(conn net.Conn, peerNodeInfo *types.NodeInfo, outbound bool, reactorsByCh map[byte]Reactor, chDescs []*ChannelDescriptor, onPeerError func(*Peer, interface{})) *Peer {
var p *Peer
onReceive := func(chId byte, msgBytes []byte) {
reactor := reactorsByCh[chId]
@@ -45,6 +73,7 @@ func newPeer(conn net.Conn, outbound bool, reactorsByCh map[byte]Reactor, chDesc
outbound: outbound,
mconn: mconn,
running: 0,
NodeInfo: peerNodeInfo,
Key: mconn.RemoteAddress.String(),
Data: NewCMap(),
}
p2p/pex_reactor.go
+2 -21
View File
@@ -101,12 +101,6 @@ func (pexR *PEXReactor) Receive(chId byte, src *Peer, msgBytes []byte) {
log.Info("Received message", "msg", msg)
switch msg.(type) {
case *pexHandshakeMessage:
network := msg.(*pexHandshakeMessage).Network
if network != pexR.sw.network {
err := fmt.Sprintf("Peer is on a different chain/network. Got %s, expected %s", network, pexR.sw.network)
pexR.sw.StopPeerForError(src, err)
}
case *pexRequestMessage:
// src requested some peers.
// TODO: prevent abuse.
@@ -219,16 +213,14 @@ func (pexR *PEXReactor) SetFireable(evsw events.Fireable) {
// Messages
const (
msgTypeRequest = byte(0x01)
msgTypeAddrs = byte(0x02)
msgTypeHandshake = byte(0x03)
msgTypeRequest = byte(0x01)
msgTypeAddrs = byte(0x02)
)
type PexMessage interface{}
var _ = binary.RegisterInterface(
struct{ PexMessage }{},
binary.ConcreteType{&pexHandshakeMessage{}, msgTypeHandshake},
binary.ConcreteType{&pexRequestMessage{}, msgTypeRequest},
binary.ConcreteType{&pexAddrsMessage{}, msgTypeAddrs},
)
@@ -241,17 +233,6 @@ func DecodeMessage(bz []byte) (msgType byte, msg PexMessage, err error) {
return
}
/*
A pexHandshakeMessage contains the network identifier.
*/
type pexHandshakeMessage struct {
Network string
}
func (m *pexHandshakeMessage) String() string {
return "[pexHandshake]"
}
/*
A pexRequestMessage requests additional peer addresses.
*/
p2p/switch.go
+20 -16
View File
@@ -8,6 +8,7 @@ import (
"time"
. "github.com/tendermint/tendermint/common"
"github.com/tendermint/tendermint/types"
)
type Reactor interface {
@@ -39,7 +40,6 @@ or more `Channels`. So while sending outgoing messages is typically performed o
incoming messages are received on the reactor.
*/
type Switch struct {
network string
listeners []Listener
reactors map[string]Reactor
chDescs []*ChannelDescriptor
@@ -47,6 +47,7 @@ type Switch struct {
peers *PeerSet
dialing *CMap
running uint32
nodeInfo *types.NodeInfo // our node info
}
var (
@@ -58,25 +59,18 @@ const (
)
func NewSwitch() *Switch {
sw := &Switch{
network: "",
reactors: make(map[string]Reactor),
chDescs: make([]*ChannelDescriptor, 0),
reactorsByCh: make(map[byte]Reactor),
peers: NewPeerSet(),
dialing: NewCMap(),
running: 0,
nodeInfo: nil,
}
return sw
}
// Not goroutine safe.
func (sw *Switch) SetNetwork(network string) {
sw.network = network
}
// Not goroutine safe.
func (sw *Switch) AddReactor(name string, reactor Reactor) Reactor {
// Validate the reactor.
@@ -109,6 +103,7 @@ func (sw *Switch) AddListener(l Listener) {
sw.listeners = append(sw.listeners, l)
}
// Not goroutine safe.
func (sw *Switch) Listeners() []Listener {
return sw.listeners
}
@@ -118,6 +113,11 @@ func (sw *Switch) IsListening() bool {
return len(sw.listeners) > 0
}
// Not goroutine safe.
func (sw *Switch) SetNodeInfo(nodeInfo *types.NodeInfo) {
sw.nodeInfo = nodeInfo
}
func (sw *Switch) Start() {
if atomic.CompareAndSwapUint32(&sw.running, 0, 1) {
// Start reactors
@@ -154,8 +154,17 @@ func (sw *Switch) Stop() {
}
}
// NOTE: This performs a blocking handshake before the peer is added.
func (sw *Switch) AddPeerWithConnection(conn net.Conn, outbound bool) (*Peer, error) {
peer := newPeer(conn, outbound, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError)
// First, perform handshake
peerNodeInfo, err := peerHandshake(conn, sw.nodeInfo)
if err != nil {
return nil, err
}
if peerNodeInfo.Network != sw.nodeInfo.Network {
return nil, fmt.Errorf("Peer is on different network %v", peerNodeInfo.Network)
}
peer := newPeer(conn, peerNodeInfo, outbound, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError)
// Add the peer to .peers
if sw.peers.Add(peer) {
@@ -177,10 +186,6 @@ func (sw *Switch) startInitPeer(peer *Peer) {
// Notify reactors
sw.doAddPeer(peer)
// Send handshake
msg := &pexHandshakeMessage{Network: sw.network}
peer.Send(PexChannel, msg)
}
func (sw *Switch) DialPeerWithAddress(addr *NetAddress) (*Peer, error) {
@@ -282,8 +287,7 @@ func (sw *Switch) listenerRoutine(l Listener) {
// New inbound connection!
peer, err := sw.AddPeerWithConnection(inConn, false)
if err != nil {
log.Info("Ignoring error from inbound connection: %v\n%v",
peer, err)
log.Info(Fmt("Ignoring error from inbound connection: %v\n%v", peer, err))
continue
}
// NOTE: We don't yet have the external address of the
p2p/switch_test.go
+14 -3
View File
@@ -8,6 +8,7 @@ import (
"github.com/tendermint/tendermint/binary"
. "github.com/tendermint/tendermint/common"
"github.com/tendermint/tendermint/types"
)
type PeerMessage struct {
@@ -73,7 +74,17 @@ func makeSwitchPair(t testing.TB, initSwitch func(*Switch) *Switch) (*Switch, *S
// Create two switches that will be interconnected.
s1 := initSwitch(NewSwitch())
s1.SetNodeInfo(&types.NodeInfo{
Moniker: "switch1",
Network: "testing",
})
s2 := initSwitch(NewSwitch())
s2.SetNodeInfo(&types.NodeInfo{
Moniker: "switch2",
Network: "testing",
})
// Start switches
s1.Start()
s2.Start()
@@ -93,7 +104,7 @@ func makeSwitchPair(t testing.TB, initSwitch func(*Switch) *Switch) (*Switch, *S
t.Fatalf("Could not get inbound connection from listener")
}
s1.AddPeerWithConnection(connIn, false)
go s1.AddPeerWithConnection(connIn, false) // AddPeer is blocking, requires handshake.
s2.AddPeerWithConnection(connOut, true)
// Wait for things to happen, peers to get added...
@@ -142,10 +153,10 @@ func TestSwitches(t *testing.T) {
// Check message on ch0
ch0Msgs := s2.Reactor("foo").(*TestReactor).msgsReceived[byte(0x00)]
if len(ch0Msgs) != 2 {
if len(ch0Msgs) != 1 {
t.Errorf("Expected to have received 1 message in ch0")
}
if !bytes.Equal(ch0Msgs[1].Bytes, binary.BinaryBytes(ch0Msg)) {
if !bytes.Equal(ch0Msgs[0].Bytes, binary.BinaryBytes(ch0Msg)) {
t.Errorf("Unexpected message bytes. Wanted: %X, Got: %X", binary.BinaryBytes(ch0Msg), ch0Msgs[0].Bytes)
}