#5287

Committed 29 Jan 2016 - 16:32 coverage decreased (-0.08%) to 75.358%

Build # #5287

Build Type

Pull #1879

circleci

Committed by

bboreham

Commit Message

Wait for plugin status via unix socket

Pull Request Pull Request #1879: Improve visibility of startup problems

Run Details

6682 of 8867 relevant lines covered (75.36%)

92936.48 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.88

/mesh/peer.go

package mesh

import (
        "crypto/rand"
        "encoding/binary"
        "fmt"
        "sort"
        "strconv"
)

func randBytes(n int) []byte {
        buf := make([]byte, n)
        _, err := rand.Read(buf)
        checkFatal(err)
        return buf
}

func randUint64() (r uint64) {
        return binary.LittleEndian.Uint64(randBytes(8))
}

func randUint16() (r uint16) {
        return binary.LittleEndian.Uint16(randBytes(2))
}

type PeerUID uint64

func randomPeerUID() PeerUID {
        for {
                uid := randUint64()
                if uid != 0 { // uid 0 is reserved for peer placeholder
                        return PeerUID(uid)
                }
        }
}

func ParsePeerUID(s string) (PeerUID, error) {
        uid, err := strconv.ParseUint(s, 10, 64)
        return PeerUID(uid), err
}

// Short IDs exist for the sake of fast datapath.  They are 12 bits,
// randomly assigned, but we detect and recover from collisions.  This
// does limit us to 4096 peers, but that should be sufficient for a
// while.
type PeerShortID uint16

const PeerShortIDBits = 12

func randomPeerShortID() PeerShortID {
        return PeerShortID(randUint16() & (1<<PeerShortIDBits - 1))
}

type PeerSummary struct {
        NameByte   []byte
        NickName   string
        UID        PeerUID
        Version    uint64
        ShortID    PeerShortID
        HasShortID bool
}

type Peer struct {
        Name PeerName
        PeerSummary
        localRefCount uint64 // maintained by Peers
        connections   map[PeerName]Connection
}

type ListOfPeers []*Peer

func (lop ListOfPeers) Len() int {
        return len(lop)
}
func (lop ListOfPeers) Swap(i, j int) {
        lop[i], lop[j] = lop[j], lop[i]
}
func (lop ListOfPeers) Less(i, j int) bool {
        return lop[i].Name < lop[j].Name
}

type ConnectionSet map[Connection]struct{}

func NewPeerFromSummary(summary PeerSummary) *Peer {
        return &Peer{
                Name:        PeerNameFromBin(summary.NameByte),
                PeerSummary: summary,
                connections: make(map[PeerName]Connection),
        }
}

func NewPeer(name PeerName, nickName string, uid PeerUID, version uint64, shortID PeerShortID) *Peer {
        return NewPeerFromSummary(PeerSummary{
                NameByte:   name.Bin(),
                NickName:   nickName,
                UID:        uid,
                Version:    version,
                ShortID:    shortID,
                HasShortID: true,
        })
}

func NewPeerPlaceholder(name PeerName) *Peer {
        return NewPeerFromSummary(PeerSummary{NameByte: name.Bin()})
}

func NewPeerFrom(peer *Peer) *Peer {
        return NewPeerFromSummary(peer.PeerSummary)
}

func (peer *Peer) String() string {
        return fmt.Sprint(peer.Name, "(", peer.NickName, ")")
}

// 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
// the message to its neighbour Y".
//
// Because currently we do not have weightings on the connections
// between peers, there is no need to use a minimum spanning tree
// algorithm. Instead we employ the simpler and cheaper breadth-first
// widening. The computation is deterministic, which ensures that when
// it is performed on the same data by different peers, they get the
// same result. This is important since otherwise we risk message loss
// or routing cycles.
//
// When the 'establishedAndSymmetric' flag is set, only connections
// that are marked as 'established' and are symmetric (i.e. where both
// sides indicate they have a connection to the other) are considered.
//
// When a non-nil stopAt peer is supplied, the widening stops when it
// reaches that peer. The boolean return indicates whether that has
// happened.
//
// 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) {
        routes := make(unicastRoutes)
        routes[peer.Name] = UnknownPeerName
        nextWorklist := []*Peer{peer}
        for len(nextWorklist) > 0 {
                worklist := nextWorklist
                sort.Sort(ListOfPeers(worklist))
                nextWorklist = []*Peer{}
                for _, curPeer := range worklist {
                        if curPeer == stopAt {
                                return true, routes
                        }
                        curPeer.ForEachConnectedPeer(establishedAndSymmetric, routes,
                                func(remotePeer *Peer) {
                                        nextWorklist = append(nextWorklist, remotePeer)
                                        remoteName := remotePeer.Name
                                        // We now know how to get to remoteName: the same
                                        // way we get to curPeer. Except, if curPeer is
                                        // the starting peer in which case we know we can
                                        // reach remoteName directly.
                                        if curPeer == peer {
                                                routes[remoteName] = remoteName
                                        } else {
                                                routes[remoteName] = routes[curPeer.Name]
                                        }
                                })
                }
        }
        return false, routes
}

func (peer *Peer) ForEachConnectedPeer(establishedAndSymmetric bool, exclude map[PeerName]PeerName, f func(*Peer)) {
        for remoteName, conn := range peer.connections {
                if establishedAndSymmetric && !conn.Established() {
                        continue
                }
                if _, found := exclude[remoteName]; found {
                        continue
                }
                remotePeer := conn.Remote()
                if remoteConn, found := remotePeer.connections[peer.Name]; !establishedAndSymmetric || (found && remoteConn.Established()) {
                        f(remotePeer)
                }
        }
}

1	package mesh
2
3	import (
4	"crypto/rand"
5	"encoding/binary"
6	"fmt"
7	"sort"
8	"strconv"
9	)
10
11	func randBytes(n int) []byte {	10,317×
12	buf := make([]byte, n)	10,317×
13	_, err := rand.Read(buf)	10,317×
14	checkFatal(err)	10,317×
15	return buf	10,317×
16	}	10,317×
17
18	func randUint64() (r uint64) {	5,190×
19	return binary.LittleEndian.Uint64(randBytes(8))	5,190×
20	}	5,190×
21
22	func randUint16() (r uint16) {	5,127×
23	return binary.LittleEndian.Uint16(randBytes(2))	5,127×
24	}	5,127×
25
26	type PeerUID uint64
27
28	func randomPeerUID() PeerUID {	5,127×
29	for {	5,127×
30	uid := randUint64()	5,127×
31	if uid != 0 { // uid 0 is reserved for peer placeholder	5,127×
32	return PeerUID(uid)	5,127×
33	}	5,127×
34	}
35	}
36
37	func ParsePeerUID(s string) (PeerUID, error) {	47×
38	uid, err := strconv.ParseUint(s, 10, 64)	47×
39	return PeerUID(uid), err	47×
40	}	47×
41
42	// Short IDs exist for the sake of fast datapath. They are 12 bits,
43	// randomly assigned, but we detect and recover from collisions. This
44	// does limit us to 4096 peers, but that should be sufficient for a
45	// while.
46	type PeerShortID uint16
47
48	const PeerShortIDBits = 12
49
50	func randomPeerShortID() PeerShortID {	5,127×
51	return PeerShortID(randUint16() & (1<<PeerShortIDBits - 1))	5,127×
52	}	5,127×
53
54	type PeerSummary struct {
55	NameByte []byte
56	NickName string
57	UID PeerUID
58	Version uint64
59	ShortID PeerShortID
60	HasShortID bool
61	}
62
63	type Peer struct {
64	Name PeerName
65	PeerSummary
66	localRefCount uint64 // maintained by Peers
67	connections map[PeerName]Connection
68	}
69
70	type ListOfPeers []*Peer
71
72	func (lop ListOfPeers) Len() int {	5,795×
73	return len(lop)	5,795×
74	}	5,795×
75	func (lop ListOfPeers) Swap(i, j int) {	14,894×
76	lop[i], lop[j] = lop[j], lop[i]	14,894×
77	}	14,894×
78	func (lop ListOfPeers) Less(i, j int) bool {	69,228×
79	return lop[i].Name < lop[j].Name	69,228×
80	}	69,228×
81
82	type ConnectionSet map[Connection]struct{}
83
84	func NewPeerFromSummary(summary PeerSummary) *Peer {	21,011×
85	return &Peer{	21,011×
86	Name: PeerNameFromBin(summary.NameByte),	21,011×
87	PeerSummary: summary,	21,011×
88	connections: make(map[PeerName]Connection),	21,011×
89	}	21,011×
90	}	21,011×
91
92	func NewPeer(name PeerName, nickName string, uid PeerUID, version uint64, shortID PeerShortID) *Peer {	13,389×
93	return NewPeerFromSummary(PeerSummary{	13,389×
94	NameByte: name.Bin(),	13,389×
95	NickName: nickName,	13,389×
96	UID: uid,	13,389×
97	Version: version,	13,389×
98	ShortID: shortID,	13,389×
99	HasShortID: true,	13,389×
100	})	13,389×
101	}	13,389×
102
UNCOV 103	func NewPeerPlaceholder(name PeerName) *Peer {	!
UNCOV 104	return NewPeerFromSummary(PeerSummary{NameByte: name.Bin()})	!
UNCOV 105	}	!
106
107	func NewPeerFrom(peer Peer) Peer {	63×
108	return NewPeerFromSummary(peer.PeerSummary)	63×
109	}	63×
110
111	func (peer *Peer) String() string {	1,463×
112	return fmt.Sprint(peer.Name, "(", peer.NickName, ")")	1,463×
113	}	1,463×
114
115	// Calculate the routing table from this peer to all peers reachable
116	// from it, returning a "next hop" map of PeerNameX -> PeerNameY,
117	// which says "in order to send a message to X, the peer should send
118	// the message to its neighbour Y".
119	//
120	// Because currently we do not have weightings on the connections
121	// between peers, there is no need to use a minimum spanning tree
122	// algorithm. Instead we employ the simpler and cheaper breadth-first
123	// widening. The computation is deterministic, which ensures that when
124	// it is performed on the same data by different peers, they get the
125	// same result. This is important since otherwise we risk message loss
126	// or routing cycles.
127	//
128	// When the 'establishedAndSymmetric' flag is set, only connections
129	// that are marked as 'established' and are symmetric (i.e. where both
130	// sides indicate they have a connection to the other) are considered.
131	//
132	// When a non-nil stopAt peer is supplied, the widening stops when it
133	// reaches that peer. The boolean return indicates whether that has
134	// happened.
135	//
136	// NB: This function should generally be invoked while holding a read
137	// lock on Peers and LocalPeer.
138	func (peer Peer) Routes(stopAt Peer, establishedAndSymmetric bool) (bool, map[PeerName]PeerName) {	3,068×
139	routes := make(unicastRoutes)	3,068×
140	routes[peer.Name] = UnknownPeerName	3,068×
141	nextWorklist := []*Peer{peer}	3,068×
142	for len(nextWorklist) > 0 {	5,795×
143	worklist := nextWorklist	5,795×
144	sort.Sort(ListOfPeers(worklist))	5,795×
145	nextWorklist = []*Peer{}	5,795×
146	for _, curPeer := range worklist {	10,295×
147	if curPeer == stopAt {	837×
148	return true, routes	837×
149	}	837×
150	curPeer.ForEachConnectedPeer(establishedAndSymmetric, routes,	9,458×
151	func(remotePeer *Peer) {	7,304×
152	nextWorklist = append(nextWorklist, remotePeer)	7,304×
153	remoteName := remotePeer.Name	7,304×
154	// We now know how to get to remoteName: the same	7,304×
155	// way we get to curPeer. Except, if curPeer is	7,304×
156	// the starting peer in which case we know we can	7,304×
157	// reach remoteName directly.	7,304×
158	if curPeer == peer {	6,404×
159	routes[remoteName] = remoteName	6,404×
160	} else {	900×
161	routes[remoteName] = routes[curPeer.Name]	900×
162	}	900×
163	})
164	}
165	}
166	return false, routes	2,231×
167	}
168
169	func (peer Peer) ForEachConnectedPeer(establishedAndSymmetric bool, exclude map[PeerName]PeerName, f func(Peer)) {	10,295×
170	for remoteName, conn := range peer.connections {	9,798×
171	if establishedAndSymmetric && !conn.Established() {	360×
172	continue	360×
173	}
174	if _, found := exclude[remoteName]; found {	1,565×
175	continue	1,565×
176	}
177	remotePeer := conn.Remote()	7,873×
178	if remoteConn, found := remotePeer.connections[peer.Name]; !establishedAndSymmetric \|\| (found && remoteConn.Established()) {	7,763×
179	f(remotePeer)	7,763×
180	}	7,763×
181	}
182	}

weaveworks / weave / #5287

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous