Merge rade/643_eliminate_route_calculcation_deadlock (PR #644)

eliminate route calculcation deadlock

Fixes #643.
This commit is contained in:
David Wragg committed 2015-05-11 17:57:53 +01:00
commit df40c5b97e
12 files changed
+174 -178

No files matched your search

+1 -1
View File
@@ -412,7 +412,7 @@ func (conn *LocalConnection) shutdown(err error) {
}
if conn.remote != nil {
conn.remote.DecrementLocalRefCount()
conn.Router.Peers.Dereference(conn.remote)
conn.Router.Ourself.DeleteConnection(conn)
}
+7 -2
View File
@@ -206,8 +206,13 @@ func (cm *ConnectionMaker) ourConnections() (PeerNameSet, map[string]struct{}, m
func (cm *ConnectionMaker) addPeerTargets(ourConnectedPeers PeerNameSet, addTarget func(string)) {
cm.peers.ForEach(func(peer *Peer) {
for conn := range peer.Connections() {
otherPeer := conn.Remote().Name
if peer == cm.ourself.Peer {
return
}
// Modifying peer.connections requires a write lock on Peers,
// and since we are holding a read lock (due to the ForEach),
// access without locking the peer is safe.
for otherPeer, conn := range peer.connections {
if otherPeer == cm.ourself.Name {
continue
}
+4 -3
View File
@@ -55,8 +55,8 @@ func (router *Router) DeleteTestChannelConnection(r *Router) {
fromPeer, _ := r.Peers.Fetch(fromName)
toPeer, _ := router.Peers.Fetch(toName)
fromPeer.DecrementLocalRefCount()
toPeer.DecrementLocalRefCount()
r.Peers.Dereference(fromPeer)
router.Peers.Dereference(toPeer)
conn, _ := router.Ourself.ConnectionTo(toName)
router.Ourself.handleDeleteConnection(conn)
@@ -79,7 +79,8 @@ func (router *Router) tp(routers ...*Router) *Peer {
p := NewPeer(r.Ourself.Peer.Name, "", r.Ourself.Peer.UID, r.Ourself.Peer.version)
connections[r.Ourself.Peer.Name] = newMockConnection(peer, p)
}
peer.SetVersionAndConnections(router.Ourself.Peer.version, connections)
peer.version = router.Ourself.Peer.version
peer.connections = connections
return peer
}
+1 -1
View File
@@ -75,7 +75,7 @@ func (conn *LocalConnection) handshake(enc *gob.Encoder, dec *gob.Decoder, accep
return fmt.Errorf("Found unknown remote name: %s at %s", name, conn.remoteTCPAddr)
}
}
if existingConn, found := conn.local.ConnectionTo(name); found && existingConn.Established() {
if existingConn, found := conn.Router.Ourself.ConnectionTo(name); found && existingConn.Established() {
return fmt.Errorf("Already have connection to %s at %s", existingConn.Remote(), existingConn.RemoteTCPAddr())
}
uid, err := strconv.ParseUint(uidStr, 10, 64)
+24 -17
View File
@@ -35,8 +35,30 @@ func (cache *MacCache) MarshalJSON() ([]byte, error) {
}
func (peers *Peers) MarshalJSON() ([]byte, error) {
var ps []*Peer
peers.ForEach(func(peer *Peer) { ps = append(ps, peer) })
type p struct {
Name string
NickName string
UID uint64
Version uint64
Connections []Connection
}
var ps []*p
peers.ForEach(func(peer *Peer) {
var connections []Connection
if peer == peers.ourself.Peer {
for conn := range peers.ourself.Connections() {
connections = append(connections, conn)
}
} else {
// Modifying peer.connections requires a write lock on
// Peers, and since we are holding a read lock (due to the
// ForEach), access without locking the peer is safe.
for _, conn := range peer.connections {
connections = append(connections, conn)
}
}
ps = append(ps, &p{peer.Name.String(), peer.NickName, peer.UID, peer.version, connections})
})
return json.Marshal(ps)
}
@@ -63,21 +85,6 @@ func (routes *Routes) MarshalJSON() ([]byte, error) {
return json.Marshal(r)
}
func (peer *Peer) MarshalJSON() ([]byte, error) {
conns := peer.Connections()
connections := make([]Connection, 0, len(conns))
for conn := range conns {
connections = append(connections, conn)
}
return json.Marshal(struct {
Name string
NickName string
UID uint64
Version uint64
Connections []Connection
}{peer.Name.String(), peer.NickName, peer.UID, peer.version, connections})
}
func (conn *RemoteConnection) MarshalJSON() ([]byte, error) {
return json.Marshal(struct {
Name string
+54 -1
View File
@@ -4,10 +4,12 @@ import (
"fmt"
"log"
"net"
"sync"
"time"
)
type LocalPeer struct {
sync.RWMutex
*Peer
router *Router
actionChan chan<- LocalPeerAction
@@ -75,6 +77,29 @@ func (peer *LocalPeer) RelayBroadcast(srcPeer *Peer, df bool, frame []byte, dec
}
}
func (peer *LocalPeer) Info() string {
peer.RLock()
defer peer.RUnlock()
return peer.Peer.Info()
}
func (peer *LocalPeer) Connections() ConnectionSet {
connections := make(ConnectionSet)
peer.RLock()
defer peer.RUnlock()
for _, conn := range peer.connections {
connections[conn] = void
}
return connections
}
func (peer *LocalPeer) ConnectionTo(name PeerName) (Connection, bool) {
peer.RLock()
defer peer.RUnlock()
conn, found := peer.connections[name]
return conn, found // yes, you really can't inline that. FFS.
}
func (peer *LocalPeer) ConnectionsTo(names []PeerName) []Connection {
conns := make([]Connection, 0, len(names))
peer.RLock()
@@ -228,6 +253,8 @@ func (peer *LocalPeer) handleDeleteConnection(conn Connection) {
peer.broadcastPeerUpdate()
}
// helpers
func (peer *LocalPeer) broadcastPeerUpdate(peers ...*Peer) {
peer.router.Routes.Recalculate()
peer.router.TopologyGossip.GossipBroadcast(NewTopologyGossipData(peer.router.Peers, append(peers, peer.Peer)...))
@@ -235,8 +262,34 @@ func (peer *LocalPeer) broadcastPeerUpdate(peers ...*Peer) {
func (peer *LocalPeer) checkConnectionLimit() error {
limit := peer.router.ConnLimit
if 0 != limit && peer.ConnectionCount() >= limit {
if 0 != limit && peer.connectionCount() >= limit {
return fmt.Errorf("Connection limit reached (%v)", limit)
}
return nil
}
func (peer *LocalPeer) addConnection(conn Connection) {
peer.Lock()
defer peer.Unlock()
peer.connections[conn.Remote().Name] = conn
peer.version++
}
func (peer *LocalPeer) deleteConnection(conn Connection) {
peer.Lock()
defer peer.Unlock()
delete(peer.connections, conn.Remote().Name)
peer.version++
}
func (peer *LocalPeer) connectionEstablished(conn Connection) {
peer.Lock()
defer peer.Unlock()
peer.version++
}
func (peer *LocalPeer) connectionCount() int {
peer.RLock()
defer peer.RUnlock()
return len(peer.connections)
}
+2 -2
View File
@@ -14,7 +14,7 @@ import (
func (peers *Peers) AddTestConnection(p *Peer) {
toPeer := NewPeer(p.Name, "", p.UID, 0)
peers.FetchWithDefault(toPeer) // Has side-effect of incrementing refcount
conn := newMockConnection(peers.ourself, toPeer)
conn := newMockConnection(peers.ourself.Peer, toPeer)
peers.ourself.addConnection(conn)
peers.ourself.connectionEstablished(conn)
}
@@ -33,7 +33,7 @@ func (peers *Peers) AddTestRemoteConnection(p1, p2 *Peer) {
func (peers *Peers) DeleteTestConnection(p *Peer) {
toName := p.Name
toPeer, _ := peers.Fetch(toName)
toPeer.DecrementLocalRefCount()
peers.Dereference(toPeer)
conn, _ := peers.ourself.ConnectionTo(toName)
peers.ourself.deleteConnection(conn)
}
+3 -91
View File
@@ -3,17 +3,15 @@ package router
import (
"fmt"
"sort"
"sync"
)
type Peer struct {
sync.RWMutex
Name PeerName
NameByte []byte
NickName string
UID uint64
version uint64
localRefCount uint64
localRefCount uint64 // maintained by Peers
connections map[PeerName]Connection
}
@@ -37,85 +35,9 @@ func (peer *Peer) String() string {
}
func (peer *Peer) Info() string {
peer.RLock()
defer peer.RUnlock()
return fmt.Sprint(peer.String(), " (v", peer.version, ") (UID ", peer.UID, ")")
}
func (peer *Peer) Version() uint64 {
peer.RLock()
defer peer.RUnlock()
return peer.version
}
func (peer *Peer) IncrementLocalRefCount() {
peer.Lock()
defer peer.Unlock()
peer.localRefCount++
}
func (peer *Peer) DecrementLocalRefCount() {
peer.Lock()
defer peer.Unlock()
peer.localRefCount--
}
func (peer *Peer) IsLocallyReferenced() bool {
peer.RLock()
defer peer.RUnlock()
return peer.localRefCount != 0
}
func (peer *Peer) ConnectionCount() int {
peer.RLock()
defer peer.RUnlock()
return len(peer.connections)
}
func (peer *Peer) ConnectionTo(name PeerName) (Connection, bool) {
peer.RLock()
defer peer.RUnlock()
conn, found := peer.connections[name]
return conn, found // yes, you really can't inline that. FFS.
}
func (peer *Peer) Connections() ConnectionSet {
connections := make(ConnectionSet)
peer.RLock()
defer peer.RUnlock()
for _, conn := range peer.connections {
connections[conn] = void
}
return connections
}
func (peer *Peer) SetVersionAndConnections(version uint64, connections map[PeerName]Connection) {
peer.Lock()
defer peer.Unlock()
peer.version = version
peer.connections = connections
}
func (peer *Peer) addConnection(conn Connection) {
peer.Lock()
defer peer.Unlock()
peer.connections[conn.Remote().Name] = conn
peer.version++
}
func (peer *Peer) deleteConnection(conn Connection) {
peer.Lock()
defer peer.Unlock()
delete(peer.connections, conn.Remote().Name)
peer.version++
}
func (peer *Peer) connectionEstablished(conn Connection) {
peer.Lock()
defer peer.Unlock()
peer.version++
}
// Calculate the routing table from this peer to all peers reachable
// from it, returning a "next hop" map of PeerNameX -> PeerNameY,
// which says "in order to send a message to X, the peer should send
@@ -137,13 +59,9 @@ func (peer *Peer) connectionEstablished(conn Connection) {
// reaches that peer. The boolean return indicates whether that has
// happened.
//
// We acquire read locks on peers as we encounter them during the
// traversal. This prevents the connectivity graph from changing
// underneath us in ways that would invalidate the result. Thus the
// answer returned may be out of date, but never inconsistent.
// NB: This function should generally be invoked while holding a read
// lock on Peers and LocalPeer.
func (peer *Peer) Routes(stopAt *Peer, establishedAndSymmetric bool) (bool, map[PeerName]PeerName) {
peer.RLock()
defer peer.RUnlock()
routes := make(map[PeerName]PeerName)
routes[peer.Name] = UnknownPeerName
nextWorklist := []*Peer{peer}
@@ -164,11 +82,7 @@ func (peer *Peer) Routes(stopAt *Peer, establishedAndSymmetric bool) (bool, map[
continue
}
remote := conn.Remote()
remote.RLock()
// DANGER holding rlock on remote, possibly taking
// rlock on remoteConn
if remoteConn, found := remote.connections[curName]; !establishedAndSymmetric || (found && remoteConn.Established()) {
defer remote.RUnlock()
nextWorklist = append(nextWorklist, remote)
// We now know how to get to remoteName: the same
// way we get to curPeer. Except, if curPeer is
@@ -179,8 +93,6 @@ func (peer *Peer) Routes(stopAt *Peer, establishedAndSymmetric bool) (bool, map[
} else {
routes[remoteName] = routes[curName]
}
} else {
remote.RUnlock()
}
}
}
+55 -43
View File
@@ -10,7 +10,7 @@ import (
type Peers struct {
sync.RWMutex
ourself *Peer
ourself *LocalPeer
table map[PeerName]*Peer
onGC func(*Peer)
}
@@ -39,7 +39,7 @@ type ConnectionSummary struct {
Established bool
}
func NewPeers(ourself *Peer, onGC func(*Peer)) *Peers {
func NewPeers(ourself *LocalPeer, onGC func(*Peer)) *Peers {
return &Peers{
ourself: ourself,
table: make(map[PeerName]*Peer),
@@ -47,20 +47,17 @@ func NewPeers(ourself *Peer, onGC func(*Peer)) *Peers {
}
func (peers *Peers) FetchWithDefault(peer *Peer) *Peer {
peers.RLock()
res, found := peers.fetchAlias(peer)
peers.RUnlock()
if found {
return res
}
peers.Lock()
defer peers.Unlock()
res, found = peers.fetchAlias(peer)
if found {
return res
if existingPeer, found := peers.table[peer.Name]; found {
if existingPeer.UID != peer.UID {
return nil
}
existingPeer.localRefCount++
return existingPeer
}
peers.table[peer.Name] = peer
peer.IncrementLocalRefCount()
peer.localRefCount++
return peer
}
@@ -71,6 +68,12 @@ func (peers *Peers) Fetch(name PeerName) (*Peer, bool) {
return peer, found // GRRR, why can't I inline this!?
}
func (peers *Peers) Dereference(peer *Peer) {
peers.Lock()
defer peers.Unlock()
peer.localRefCount--
}
func (peers *Peers) ForEach(fun func(*Peer)) {
peers.RLock()
defer peers.RUnlock()
@@ -125,19 +128,21 @@ func (peers *Peers) Names() PeerNameSet {
}
func (peers *Peers) EncodePeers(names PeerNameSet) []byte {
buf := new(bytes.Buffer)
enc := gob.NewEncoder(buf)
peers.RLock()
defer peers.RUnlock()
peerList := make([]*Peer, 0, len(names))
for name := range names {
if peer, found := peers.table[name]; found {
if peer == peers.ourself.Peer {
peers.ourself.Peer.Encode(enc)
} else {
peer.Encode(enc)
}
peerList = append(peerList, peer)
}
}
peers.RUnlock() // release lock so we don't hold it while encoding
buf := new(bytes.Buffer)
enc := gob.NewEncoder(buf)
for _, peer := range peerList {
peer.encode(enc)
}
return buf.Bytes()
}
@@ -149,35 +154,39 @@ func (peers *Peers) GarbageCollect() []*Peer {
func (peers *Peers) String() string {
var buf bytes.Buffer
printConnection := func(conn Connection) {
established := ""
if !conn.Established() {
established = " (unestablished)"
}
fmt.Fprintf(&buf, " -> %s [%v%s]\n", conn.Remote(), conn.RemoteTCPAddr(), established)
}
peers.ForEach(func(peer *Peer) {
fmt.Fprintln(&buf, peer.Info())
for conn := range peer.Connections() {
established := ""
if !conn.Established() {
established = " (unestablished)"
if peer == peers.ourself.Peer {
fmt.Fprintln(&buf, peers.ourself.Info())
for conn := range peers.ourself.Connections() {
printConnection(conn)
}
} else {
fmt.Fprintln(&buf, peer.Info())
// Modifying peer.connections requires a write lock on
// Peers, and since we are holding a read lock (due to the
// ForEach), access without locking the peer is safe.
for _, conn := range peer.connections {
printConnection(conn)
}
fmt.Fprintf(&buf, " -> %s [%v%s]\n", conn.Remote(), conn.RemoteTCPAddr(), established)
}
})
return buf.String()
}
func (peers *Peers) fetchAlias(peer *Peer) (*Peer, bool) {
if existingPeer, found := peers.table[peer.Name]; found {
if existingPeer.UID != peer.UID {
return nil, true
}
existingPeer.IncrementLocalRefCount()
return existingPeer, true
}
return nil, false
}
func (peers *Peers) garbageCollect() []*Peer {
removed := []*Peer{}
peers.ourself.RLock()
_, reached := peers.ourself.Routes(nil, false)
peers.ourself.RUnlock()
for name, peer := range peers.table {
if _, found := reached[peer.Name]; !found && !peer.IsLocallyReferenced() {
if _, found := reached[peer.Name]; !found && peer.localRefCount == 0 {
delete(peers.table, name)
peers.onGC(peer)
removed = append(removed, peer)
@@ -250,7 +259,7 @@ func (peers *Peers) applyUpdate(decodedUpdate []*Peer, decodedConns [][]Connecti
// guaranteed to find peer in the peers.table
peer := peers.table[name]
if peer != newPeer &&
(peer == peers.ourself || peer.Version() >= newPeer.Version()) {
(peer == peers.ourself.Peer || peer.version >= newPeer.version) {
// Nobody but us updates us. And if we know more about a
// peer than what's in the the update, we ignore the
// latter.
@@ -266,17 +275,14 @@ func (peers *Peers) applyUpdate(decodedUpdate []*Peer, decodedConns [][]Connecti
// for an update would be someone else calling
// router.Peers.ApplyUpdate. But ApplyUpdate takes the Lock on
// the router.Peers, so there can be no race here.
conns := makeConnsMap(peer, connSummaries, peers.table)
peer.SetVersionAndConnections(newPeer.Version(), conns)
peer.version = newPeer.version
peer.connections = makeConnsMap(peer, connSummaries, peers.table)
newUpdate[name] = peer
}
return newUpdate
}
func (peer *Peer) encode(enc *gob.Encoder) {
peer.RLock()
defer peer.RUnlock()
func (peer *Peer) Encode(enc *gob.Encoder) {
checkFatal(enc.Encode(PeerSummary{
peer.NameByte,
peer.NickName,
@@ -297,6 +303,12 @@ func (peer *Peer) encode(enc *gob.Encoder) {
checkFatal(enc.Encode(connSummaries))
}
func (peer *LocalPeer) Encode(enc *gob.Encoder) {
peer.RLock()
defer peer.RUnlock()
peer.Peer.Encode(enc)
}
func decodePeer(dec *gob.Decoder) (peerSummary PeerSummary, connSummaries []ConnectionSummary, err error) {
if err = dec.Decode(&peerSummary); err != nil {
return
+5 -5
View File
@@ -17,10 +17,10 @@ import (
// - non-gc of peers that are only referenced locally
func newNode(name PeerName) (*Peer, *Peers) {
peer := NewPeer(name, "", 0, 0)
peer := NewLocalPeer(name, "", nil)
peers := NewPeers(peer, func(*Peer) {})
peers.FetchWithDefault(peer)
return peer, peers
peers.FetchWithDefault(peer.Peer)
return peer.Peer, peers
}
// Check that ApplyUpdate copies the whole topology from peers
@@ -29,7 +29,7 @@ func checkApplyUpdate(t *testing.T, peers *Peers) {
// We need a new node outside of the network, with a connection
// into it.
_, testBedPeers := newNode(dummyName)
testBedPeers.AddTestConnection(peers.ourself)
testBedPeers.AddTestConnection(peers.ourself.Peer)
testBedPeers.ApplyUpdate(peers.EncodePeers(peers.Names()))
checkTopologyPeers(t, true, testBedPeers.allPeersExcept(dummyName), peers.allPeers()...)
@@ -52,7 +52,7 @@ func TestPeersEncoding(t *testing.T) {
case 0:
from, to := rand.Intn(numNodes), rand.Intn(numNodes)
if from != to {
if _, found := peer[from].ConnectionTo(peer[to].Name); !found {
if _, found := peer[from].connections[peer[to].Name]; !found {
ps[from].AddTestConnection(peer[to])
conns = append(conns, struct{ from, to int }{from, to})
checkApplyUpdate(t, ps[from])
+2 -2
View File
@@ -64,9 +64,9 @@ func NewRouter(config RouterConfig, name PeerName, nickName string) *Router {
}
router.Ourself = NewLocalPeer(name, nickName, router)
router.Macs = NewMacCache(macMaxAge, onMacExpiry)
router.Peers = NewPeers(router.Ourself.Peer, onPeerGC)
router.Peers = NewPeers(router.Ourself, onPeerGC)
router.Peers.FetchWithDefault(router.Ourself.Peer)
router.Routes = NewRoutes(router.Ourself.Peer, router.Peers)
router.Routes = NewRoutes(router.Ourself, router.Peers)
router.ConnectionMaker = NewConnectionMaker(router.Ourself, router.Peers, router.Port)
router.TopologyGossip = router.NewGossip("topology", router)
return router
+16 -10
View File
@@ -9,7 +9,7 @@ import (
type Routes struct {
sync.RWMutex
ourself *Peer
ourself *LocalPeer
peers *Peers
unicast map[PeerName]PeerName
unicastAll map[PeerName]PeerName // [1]
@@ -21,7 +21,7 @@ type Routes struct {
// symmetric ones
}
func NewRoutes(ourself *Peer, peers *Peers) *Routes {
func NewRoutes(ourself *LocalPeer, peers *Peers) *Routes {
routes := &Routes{
ourself: ourself,
peers: peers,
@@ -166,12 +166,17 @@ func (routes *Routes) run(recalculate <-chan *struct{}, wait <-chan chan struct{
}
func (routes *Routes) calculate() {
routes.peers.RLock()
routes.ourself.RLock()
var (
unicast = routes.calculateUnicast(true)
unicastAll = routes.calculateUnicast(false)
broadcast = routes.calculateBroadcast(true)
broadcastAll = routes.calculateBroadcast(false)
)
routes.ourself.RUnlock()
routes.peers.RUnlock()
routes.Lock()
routes.unicast = unicast
routes.unicastAll = unicastAll
@@ -214,16 +219,13 @@ func (routes *Routes) calculateUnicast(establishedAndSymmetric bool) map[PeerNam
// where <= is the subset relationship on keys of the returned map.
func (routes *Routes) calculateBroadcast(establishedAndSymmetric bool) map[PeerName][]PeerName {
broadcast := make(map[PeerName][]PeerName)
ourself := routes.ourself
ourConnections := ourself.Connections()
routes.peers.ForEach(func(peer *Peer) {
for _, peer := range routes.peers.table {
hops := []PeerName{}
if found, reached := peer.Routes(ourself, establishedAndSymmetric); found {
if found, reached := peer.Routes(routes.ourself.Peer, establishedAndSymmetric); found {
// This is rather similar to the inner loop on
// peer.Routes(...); the main difference is in the
// locking.
for conn := range ourConnections {
for _, conn := range routes.ourself.connections {
if establishedAndSymmetric && !conn.Established() {
continue
}
@@ -231,12 +233,16 @@ func (routes *Routes) calculateBroadcast(establishedAndSymmetric bool) map[PeerN
if _, found := reached[remoteName]; found {
continue
}
if remoteConn, found := conn.Remote().ConnectionTo(ourself.Name); !establishedAndSymmetric || (found && remoteConn.Established()) {
// conn.Remote() cannot by ourself. Modifying
// peer.connections requires a write lock on Peers,
// and since we are holding a read lock (due to the
// ForEach), access without locking the peer is safe.
if remoteConn, found := conn.Remote().connections[routes.ourself.Name]; !establishedAndSymmetric || (found && remoteConn.Established()) {
hops = append(hops, remoteName)
}
}
}
broadcast[peer.Name] = hops
})
}
return broadcast
}