Add an 'outbound' field on each connection,
and only attempt new connections to the remote side of outbound connections.
This commit is contained in:
1 parent
6f89ead9bd
commit
89fce7f98a
8 files changed
+26
-14
No files matched your search
@@ -16,6 +16,7 @@ type Connection interface {
|
||||
BreakTie(Connection) ConnectionTieBreak
|
||||
RemoteTCPAddr() string
|
||||
Established() bool
|
||||
Outbound() bool
|
||||
Shutdown(error)
|
||||
Log(args ...interface{})
|
||||
}
|
||||
@@ -33,6 +34,7 @@ type RemoteConnection struct {
|
||||
remote *Peer
|
||||
remoteTCPAddr string
|
||||
established bool
|
||||
outbound bool
|
||||
}
|
||||
|
||||
type LocalConnection struct {
|
||||
@@ -66,12 +68,13 @@ type ConnectionInteraction struct {
|
||||
payload interface{}
|
||||
}
|
||||
|
||||
func NewRemoteConnection(from, to *Peer, tcpAddr string, established bool) *RemoteConnection {
|
||||
func NewRemoteConnection(from, to *Peer, tcpAddr string, established bool, outbound bool) *RemoteConnection {
|
||||
return &RemoteConnection{
|
||||
local: from,
|
||||
remote: to,
|
||||
remoteTCPAddr: tcpAddr,
|
||||
established: established}
|
||||
established: established,
|
||||
outbound: outbound}
|
||||
}
|
||||
|
||||
func (conn *RemoteConnection) Local() *Peer {
|
||||
@@ -94,6 +97,10 @@ func (conn *RemoteConnection) Established() bool {
|
||||
return conn.established
|
||||
}
|
||||
|
||||
func (conn *RemoteConnection) Outbound() bool {
|
||||
return conn.outbound
|
||||
}
|
||||
|
||||
func (conn *RemoteConnection) Shutdown(error) {
|
||||
}
|
||||
|
||||
|
||||
@@ -140,7 +140,7 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections() time.Duration {
|
||||
// aren't
|
||||
cm.peers.ForEach(func(name PeerName, peer *Peer) {
|
||||
peer.ForEachConnection(func(otherPeer PeerName, conn Connection) {
|
||||
if otherPeer == cm.ourself.Name || ourConnectedPeers[otherPeer] {
|
||||
if otherPeer == cm.ourself.Name || ourConnectedPeers[otherPeer] || !conn.Outbound() {
|
||||
return
|
||||
}
|
||||
address := conn.RemoteTCPAddr()
|
||||
|
||||
@@ -41,7 +41,7 @@ func (router *Router) AddTestChannelConnection(r *Router) {
|
||||
r.Peers.FetchWithDefault(fromPeer) // Has side-effect of incrementing refcount
|
||||
router.Peers.FetchWithDefault(toPeer) //
|
||||
|
||||
conn := &mockChannelConnection{RemoteConnection{router.Ourself.Peer, toPeer, "", false}, r}
|
||||
conn := &mockChannelConnection{RemoteConnection{router.Ourself.Peer, toPeer, "", false, true}, r}
|
||||
router.Ourself.handleAddConnection(conn)
|
||||
router.Ourself.handleConnectionEstablished(conn)
|
||||
}
|
||||
|
||||
@@ -111,7 +111,7 @@ func (peer *LocalPeer) CreateConnection(peerAddr string, acceptNewPeer bool) err
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
connRemote := NewRemoteConnection(peer.Peer, nil, tcpConn.RemoteAddr().String(), false)
|
||||
connRemote := NewRemoteConnection(peer.Peer, nil, tcpConn.RemoteAddr().String(), false, true)
|
||||
connLocal := NewLocalConnection(connRemote, tcpConn, udpAddr, peer.Router)
|
||||
connLocal.Start(acceptNewPeer)
|
||||
return nil
|
||||
|
||||
@@ -27,7 +27,7 @@ func (peers *Peers) AddTestRemoteConnection(p1, p2 *Peer) {
|
||||
toName := p2.Name
|
||||
toPeer := NewPeer(toName, "", p2.UID, 0)
|
||||
toPeer = peers.FetchWithDefault(toPeer)
|
||||
peers.ourself.addConnection(&RemoteConnection{fromPeer, toPeer, "", false})
|
||||
peers.ourself.addConnection(&RemoteConnection{fromPeer, toPeer, "", false, false})
|
||||
}
|
||||
|
||||
func (peers *Peers) DeleteTestConnection(p *Peer) {
|
||||
@@ -44,7 +44,7 @@ func (peers *Peers) DeleteTestConnection(p *Peer) {
|
||||
// from what is created by the real code.
|
||||
func newMockConnection(from, to *Peer) Connection {
|
||||
type mockConnection struct{ RemoteConnection }
|
||||
return &mockConnection{RemoteConnection{from, to, "", false}}
|
||||
return &mockConnection{RemoteConnection{from, to, "", false, false}}
|
||||
}
|
||||
|
||||
func checkEqualConns(t *testing.T, ourName PeerName, got, wanted map[PeerName]Connection) {
|
||||
|
||||
+10
-5
@@ -199,7 +199,7 @@ func (peers *Peers) decodeUpdate(update []byte) (newPeers map[PeerName]*Peer, de
|
||||
}
|
||||
|
||||
for _, connsBuf := range decodedConns {
|
||||
decErr := connsIterator(connsBuf, func(remoteNameByte []byte, _ string, _ bool) {
|
||||
decErr := connsIterator(connsBuf, func(remoteNameByte []byte, _ string, _, _ bool) {
|
||||
remoteName := PeerNameFromBin(remoteNameByte)
|
||||
if _, found := newPeers[remoteName]; found {
|
||||
return
|
||||
@@ -268,6 +268,7 @@ func (peer *Peer) encode(enc *gob.Encoder) {
|
||||
checkFatal(connsEnc.Encode(conn.RemoteTCPAddr()))
|
||||
// DANGER holding rlock on peer, going to take rlock on conn
|
||||
checkFatal(connsEnc.Encode(conn.Established()))
|
||||
checkFatal(connsEnc.Encode(conn.Outbound()))
|
||||
}
|
||||
checkFatal(enc.Encode(connsBuf.Bytes()))
|
||||
}
|
||||
@@ -291,7 +292,7 @@ func decodePeerNoConns(dec *gob.Decoder) (nameByte []byte, nickName string, uid
|
||||
return
|
||||
}
|
||||
|
||||
func connsIterator(input []byte, fun func([]byte, string, bool)) error {
|
||||
func connsIterator(input []byte, fun func([]byte, string, bool, bool)) error {
|
||||
buf := new(bytes.Buffer)
|
||||
buf.Write(input)
|
||||
dec := gob.NewDecoder(buf)
|
||||
@@ -308,16 +309,20 @@ func connsIterator(input []byte, fun func([]byte, string, bool)) error {
|
||||
if err := dec.Decode(&established); err != nil {
|
||||
return err
|
||||
}
|
||||
fun(nameByte, foundAt, established)
|
||||
var outbound bool
|
||||
if err := dec.Decode(&outbound); err != nil {
|
||||
return err
|
||||
}
|
||||
fun(nameByte, foundAt, established, outbound)
|
||||
}
|
||||
}
|
||||
|
||||
func readConnsMap(peer *Peer, buf []byte, table map[PeerName]*Peer) map[PeerName]Connection {
|
||||
conns := make(map[PeerName]Connection)
|
||||
if err := connsIterator(buf, func(nameByte []byte, remoteTCPAddr string, established bool) {
|
||||
if err := connsIterator(buf, func(nameByte []byte, remoteTCPAddr string, established bool, outbound bool) {
|
||||
name := PeerNameFromBin(nameByte)
|
||||
remotePeer := table[name]
|
||||
conn := NewRemoteConnection(peer, remotePeer, remoteTCPAddr, established)
|
||||
conn := NewRemoteConnection(peer, remotePeer, remoteTCPAddr, established, outbound)
|
||||
conns[name] = conn
|
||||
}); err != io.EOF {
|
||||
// this should never happen since we've already successfully
|
||||
|
||||
+1
-1
@@ -2,7 +2,7 @@ package router
|
||||
|
||||
const (
|
||||
Protocol = "weave"
|
||||
ProtocolVersion = 13
|
||||
ProtocolVersion = 14
|
||||
)
|
||||
|
||||
type ProtocolTag byte
|
||||
|
||||
+1
-1
@@ -192,7 +192,7 @@ func (router *Router) acceptTCP(tcpConn *net.TCPConn) {
|
||||
// start.
|
||||
remoteAddrStr := tcpConn.RemoteAddr().String()
|
||||
log.Printf("->[%s] connection accepted\n", remoteAddrStr)
|
||||
connRemote := NewRemoteConnection(router.Ourself.Peer, nil, remoteAddrStr, false)
|
||||
connRemote := NewRemoteConnection(router.Ourself.Peer, nil, remoteAddrStr, false, false)
|
||||
connLocal := NewLocalConnection(connRemote, tcpConn, nil, router)
|
||||
connLocal.Start(true)
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user