refactor: rename PeerCache to Peers
It was never much of a cache, and that aspect of it is certainly not its defining characteristic.
This commit is contained in:
1 parent
e8f836ad85
commit
2b2f47c031
4 files changed
+66
-67
No files matched your search
+12
-13
@@ -96,19 +96,18 @@ Router:
|
||||
directly connected neighbours via TCP. This network update message
|
||||
contains the relevant changes to the network topology due to the
|
||||
connection change. When such a message is received by a TCP
|
||||
receiver threads, they invoke the peers cache to apply the update
|
||||
to the local model of the network topology. This may fail for a
|
||||
number of reasons (for example the update may contain reference to
|
||||
peers that we have no prior knowledge of. In this case, we ignore
|
||||
the update and send back to the peer from whom we received the
|
||||
update a request for the complete network topology), or it may
|
||||
apply and elicit some changes to our model. If it does elicit some
|
||||
changes then we send out an updated update message to all our
|
||||
peers. In this way the changes are passed quickly from peer to
|
||||
peer based on the established connections and stop being sent once
|
||||
a received update causes no changes to the peer's topology
|
||||
model. Changes are additive and care is taken to ensure that no
|
||||
loops can occur.
|
||||
receiver threads, they apply the update to the local model of the
|
||||
network topology. This may fail for a number of reasons (for
|
||||
example the update may contain reference to peers that we have no
|
||||
prior knowledge of. In this case, we ignore the update and send
|
||||
back to the peer from whom we received the update a request for
|
||||
the complete network topology), or it may apply and elicit some
|
||||
changes to our model. If it does elicit some changes then we send
|
||||
out an updated update message to all our peers. In this way the
|
||||
changes are passed quickly from peer to peer based on the
|
||||
established connections and stop being sent once a received update
|
||||
causes no changes to the peer's topology model. Changes are
|
||||
additive and care is taken to ensure that no loops can occur.
|
||||
|
||||
|
||||
Routing
|
||||
|
||||
@@ -8,55 +8,55 @@ import (
|
||||
"sync"
|
||||
)
|
||||
|
||||
type PeerCache struct {
|
||||
type Peers struct {
|
||||
sync.RWMutex
|
||||
table map[PeerName]*Peer
|
||||
onGC func(*Peer)
|
||||
}
|
||||
|
||||
func NewPeerCache(onGC func(*Peer)) *PeerCache {
|
||||
return &PeerCache{table: make(map[PeerName]*Peer), onGC: onGC}
|
||||
func NewPeers(onGC func(*Peer)) *Peers {
|
||||
return &Peers{table: make(map[PeerName]*Peer), onGC: onGC}
|
||||
}
|
||||
|
||||
func (cache *PeerCache) FetchWithDefault(peer *Peer) *Peer {
|
||||
cache.RLock()
|
||||
res, found := cache.fetchAlias(peer)
|
||||
cache.RUnlock()
|
||||
func (peers *Peers) FetchWithDefault(peer *Peer) *Peer {
|
||||
peers.RLock()
|
||||
res, found := peers.fetchAlias(peer)
|
||||
peers.RUnlock()
|
||||
if found {
|
||||
return res
|
||||
}
|
||||
cache.Lock()
|
||||
defer cache.Unlock()
|
||||
res, found = cache.fetchAlias(peer)
|
||||
peers.Lock()
|
||||
defer peers.Unlock()
|
||||
res, found = peers.fetchAlias(peer)
|
||||
if found {
|
||||
return res
|
||||
}
|
||||
cache.table[peer.Name] = peer
|
||||
peers.table[peer.Name] = peer
|
||||
peer.IncrementLocalRefCount()
|
||||
return peer
|
||||
}
|
||||
|
||||
func (cache *PeerCache) Fetch(name PeerName) (*Peer, bool) {
|
||||
cache.RLock()
|
||||
defer cache.RUnlock()
|
||||
peer, found := cache.table[name]
|
||||
func (peers *Peers) Fetch(name PeerName) (*Peer, bool) {
|
||||
peers.RLock()
|
||||
defer peers.RUnlock()
|
||||
peer, found := peers.table[name]
|
||||
return peer, found // GRRR, why can't I inline this!?
|
||||
}
|
||||
|
||||
func (cache *PeerCache) ForEach(fun func(PeerName, *Peer)) {
|
||||
cache.RLock()
|
||||
defer cache.RUnlock()
|
||||
for name, peer := range cache.table {
|
||||
func (peers *Peers) ForEach(fun func(PeerName, *Peer)) {
|
||||
peers.RLock()
|
||||
defer peers.RUnlock()
|
||||
for name, peer := range peers.table {
|
||||
fun(name, peer)
|
||||
}
|
||||
}
|
||||
|
||||
func (cache *PeerCache) ApplyUpdate(update []byte, router *Router) ([]byte, error) {
|
||||
cache.Lock()
|
||||
func (peers *Peers) ApplyUpdate(update []byte, router *Router) ([]byte, error) {
|
||||
peers.Lock()
|
||||
|
||||
newPeers, decodedUpdate, decodedConns, err := cache.decodeUpdate(update, router)
|
||||
newPeers, decodedUpdate, decodedConns, err := peers.decodeUpdate(update, router)
|
||||
if err != nil {
|
||||
cache.Unlock()
|
||||
peers.Unlock()
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -64,18 +64,18 @@ func (cache *PeerCache) ApplyUpdate(update []byte, router *Router) ([]byte, erro
|
||||
// have no knowledge of. We can now apply the update. Start by
|
||||
// adding in any new peers into the cache.
|
||||
for name, newPeer := range newPeers {
|
||||
cache.table[name] = newPeer
|
||||
peers.table[name] = newPeer
|
||||
}
|
||||
|
||||
// Now apply the updates
|
||||
newUpdate := cache.applyUpdate(decodedUpdate, decodedConns, router)
|
||||
newUpdate := peers.applyUpdate(decodedUpdate, decodedConns, router)
|
||||
|
||||
for _, peerRemoved := range cache.garbageCollect(router.Ourself) {
|
||||
for _, peerRemoved := range peers.garbageCollect(router.Ourself) {
|
||||
delete(newUpdate, peerRemoved.Name)
|
||||
}
|
||||
|
||||
// Don't need to hold cache lock any longer
|
||||
cache.Unlock()
|
||||
// Don't need to hold peers lock any longer
|
||||
peers.Unlock()
|
||||
|
||||
if len(newUpdate) > 0 {
|
||||
router.ConnectionMaker.Refresh()
|
||||
@@ -84,10 +84,10 @@ func (cache *PeerCache) ApplyUpdate(update []byte, router *Router) ([]byte, erro
|
||||
return encodePeersMap(newUpdate), nil
|
||||
}
|
||||
|
||||
func (cache *PeerCache) EncodeAllPeers() []byte {
|
||||
cache.RLock()
|
||||
defer cache.RUnlock()
|
||||
return encodePeersMap(cache.table)
|
||||
func (peers *Peers) EncodeAllPeers() []byte {
|
||||
peers.RLock()
|
||||
defer peers.RUnlock()
|
||||
return encodePeersMap(peers.table)
|
||||
}
|
||||
|
||||
func EncodePeers(peers ...*Peer) []byte {
|
||||
@@ -99,15 +99,15 @@ func EncodePeers(peers ...*Peer) []byte {
|
||||
return buf.Bytes()
|
||||
}
|
||||
|
||||
func (cache *PeerCache) GarbageCollect(router *Router) []*Peer {
|
||||
cache.Lock()
|
||||
defer cache.Unlock()
|
||||
return cache.garbageCollect(router.Ourself)
|
||||
func (peers *Peers) GarbageCollect(router *Router) []*Peer {
|
||||
peers.Lock()
|
||||
defer peers.Unlock()
|
||||
return peers.garbageCollect(router.Ourself)
|
||||
}
|
||||
|
||||
func (cache *PeerCache) String() string {
|
||||
func (peers *Peers) String() string {
|
||||
var buf bytes.Buffer
|
||||
cache.ForEach(func(name PeerName, peer *Peer) {
|
||||
peers.ForEach(func(name PeerName, peer *Peer) {
|
||||
buf.WriteString(fmt.Sprint(peer, "\n"))
|
||||
peer.ForEachConnection(func(remoteName PeerName, conn Connection) {
|
||||
buf.WriteString(fmt.Sprintf(" -> %v [%v]\n", remoteName, conn.RemoteTCPAddr()))
|
||||
@@ -116,8 +116,8 @@ func (cache *PeerCache) String() string {
|
||||
return buf.String()
|
||||
}
|
||||
|
||||
func (cache *PeerCache) fetchAlias(peer *Peer) (*Peer, bool) {
|
||||
if existingPeer, found := cache.table[peer.Name]; found {
|
||||
func (peers *Peers) fetchAlias(peer *Peer) (*Peer, bool) {
|
||||
if existingPeer, found := peers.table[peer.Name]; found {
|
||||
if existingPeer.UID == peer.UID {
|
||||
existingPeer.IncrementLocalRefCount()
|
||||
return existingPeer, true
|
||||
@@ -128,13 +128,13 @@ func (cache *PeerCache) fetchAlias(peer *Peer) (*Peer, bool) {
|
||||
return nil, false
|
||||
}
|
||||
|
||||
func (cache *PeerCache) garbageCollect(ourself *Peer) []*Peer {
|
||||
func (peers *Peers) garbageCollect(ourself *Peer) []*Peer {
|
||||
removed := []*Peer{}
|
||||
for name, peer := range cache.table {
|
||||
for name, peer := range peers.table {
|
||||
found, _ := ourself.Routes(peer, false)
|
||||
if !found && !peer.IsLocallyReferenced() {
|
||||
cache.onGC(peer)
|
||||
delete(cache.table, name)
|
||||
peers.onGC(peer)
|
||||
delete(peers.table, name)
|
||||
ourself.Router.Macs.Delete(peer)
|
||||
removed = append(removed, peer)
|
||||
}
|
||||
@@ -151,7 +151,7 @@ func encodePeersMap(peers map[PeerName]*Peer) []byte {
|
||||
return buf.Bytes()
|
||||
}
|
||||
|
||||
func (cache *PeerCache) decodeUpdate(update []byte, router *Router) (newPeers map[PeerName]*Peer, decodedUpdate []*Peer, decodedConns [][]byte, err error) {
|
||||
func (peers *Peers) decodeUpdate(update []byte, router *Router) (newPeers map[PeerName]*Peer, decodedUpdate []*Peer, decodedConns [][]byte, err error) {
|
||||
newPeers = make(map[PeerName]*Peer)
|
||||
decodedUpdate = []*Peer{}
|
||||
decodedConns = [][]byte{}
|
||||
@@ -172,7 +172,7 @@ func (cache *PeerCache) decodeUpdate(update []byte, router *Router) (newPeers ma
|
||||
newPeer := NewPeer(name, uid, version, router)
|
||||
decodedUpdate = append(decodedUpdate, newPeer)
|
||||
decodedConns = append(decodedConns, connsBuf)
|
||||
existingPeer, found := cache.table[name]
|
||||
existingPeer, found := peers.table[name]
|
||||
if !found {
|
||||
newPeers[name] = newPeer
|
||||
} else if existingPeer.UID != newPeer.UID {
|
||||
@@ -188,7 +188,7 @@ func (cache *PeerCache) decodeUpdate(update []byte, router *Router) (newPeers ma
|
||||
if _, found := newPeers[remoteName]; found {
|
||||
return
|
||||
}
|
||||
if _, found := cache.table[remoteName]; found {
|
||||
if _, found := peers.table[remoteName]; found {
|
||||
return
|
||||
}
|
||||
// Update refers to a peer which we have no knowledge
|
||||
@@ -203,13 +203,13 @@ func (cache *PeerCache) decodeUpdate(update []byte, router *Router) (newPeers ma
|
||||
return
|
||||
}
|
||||
|
||||
func (cache *PeerCache) applyUpdate(decodedUpdate []*Peer, decodedConns [][]byte, router *Router) map[PeerName]*Peer {
|
||||
func (peers *Peers) applyUpdate(decodedUpdate []*Peer, decodedConns [][]byte, router *Router) map[PeerName]*Peer {
|
||||
newUpdate := make(map[PeerName]*Peer)
|
||||
for idx, newPeer := range decodedUpdate {
|
||||
connsBuf := decodedConns[idx]
|
||||
name := newPeer.Name
|
||||
// guaranteed to find peer in the cache.table
|
||||
peer := cache.table[name]
|
||||
// guaranteed to find peer in the peers.table
|
||||
peer := peers.table[name]
|
||||
if peer != newPeer {
|
||||
if peer.Version() > newPeer.Version() {
|
||||
// we know more about this one than the update. If
|
||||
@@ -237,8 +237,8 @@ func (cache *PeerCache) applyUpdate(decodedUpdate []*Peer, decodedConns [][]byte
|
||||
// No - we know that peer is not ourself, so the only prospect
|
||||
// for an update would be someone else calling
|
||||
// router.Peers.ApplyUpdate. But ApplyUpdate takes the Lock on
|
||||
// the peers cache, so there can be no race here.
|
||||
conns := readConnsMap(peer, connsBuf, cache.table)
|
||||
// the router.Peers, so there can be no race here.
|
||||
conns := readConnsMap(peer, connsBuf, peers.table)
|
||||
peer.SetVersionAndConnections(newPeer.Version(), conns)
|
||||
newUpdate[name] = peer
|
||||
}
|
||||
+1
-1
@@ -27,7 +27,7 @@ func NewRouter(iface *net.Interface, name PeerName, password []byte, connLimit i
|
||||
router := &Router{
|
||||
Iface: iface,
|
||||
Macs: NewMacCache(macMaxAge, onMacExpiry),
|
||||
Peers: NewPeerCache(onPeerGC),
|
||||
Peers: NewPeers(onPeerGC),
|
||||
ConnLimit: connLimit,
|
||||
BufSz: bufSz,
|
||||
LogFrame: logFrame}
|
||||
|
||||
+1
-1
@@ -15,7 +15,7 @@ type Router struct {
|
||||
Ourself *Peer
|
||||
Iface *net.Interface
|
||||
Macs *MacCache
|
||||
Peers *PeerCache
|
||||
Peers *Peers
|
||||
UDPListener *net.UDPConn
|
||||
Routes *Routes
|
||||
ConnectionMaker *ConnectionMaker
|
||||
|
||||
Reference in new issue
Block a user