5
0
mirror of https://github.com/cwinfo/yggdrasil-go.git synced 2024-12-22 19:55:39 +00:00

possibly fix deadlock from race in peer linkloop goroutine, add some related debug code to the admin

This commit is contained in:
Arceliar 2018-02-07 17:48:30 -06:00
parent 63aadf6e88
commit e9adf327b0
4 changed files with 47 additions and 39 deletions

View File

@ -41,11 +41,14 @@ func (a *admin) init(c *Core, listenaddr string) {
*out = a.getResponse_dot()
})
a.addHandler("getSelf", nil, func(out *[]byte, _ ...string) {
*out = []byte(a.printInfos(a.getData_getPeers()))
*out = []byte(a.printInfos([]admin_nodeInfo{*a.getData_getSelf()}))
})
a.addHandler("getPeers", nil, func(out *[]byte, _ ...string) {
*out = []byte(a.printInfos(a.getData_getPeers()))
})
a.addHandler("getSwitchPeers", nil, func(out *[]byte, _ ...string) {
*out = []byte(a.printInfos(a.getData_getSwitchPeers()))
})
a.addHandler("getDHT", nil, func(out *[]byte, _ ...string) {
*out = []byte(a.printInfos(a.getData_getDHT()))
})
@ -152,6 +155,26 @@ func (a *admin) getData_getSelf() *admin_nodeInfo {
}
func (a *admin) getData_getPeers() []admin_nodeInfo {
ports := a.core.peers.ports.Load().(map[switchPort]*peer)
var peerInfos []admin_nodeInfo
var ps []switchPort
for port := range ports {
ps = append(ps, port)
}
sort.Slice(ps, func(i, j int) bool { return ps[i] < ps[j] })
for _, port := range ps {
p := ports[port]
addr := *address_addrForNodeID(getNodeID(&p.box))
info := admin_nodeInfo{
{"IP", net.IP(addr[:]).String()},
{"port", fmt.Sprint(port)},
}
peerInfos = append(peerInfos, info)
}
return peerInfos
}
func (a *admin) getData_getSwitchPeers() []admin_nodeInfo {
var peerInfos []admin_nodeInfo
table := a.core.switchTable.table.Load().(lookupTable)
peers := a.core.peers.ports.Load().(map[switchPort]*peer)
@ -211,7 +234,7 @@ func (a *admin) getData_getSessions() []admin_nodeInfo {
func (a *admin) getResponse_dot() []byte {
self := a.getData_getSelf().asMap()
myAddr := self["IP"]
peers := a.getData_getPeers()
peers := a.getData_getSwitchPeers()
dht := a.getData_getDHT()
sessions := a.getData_getSessions()
// Map of coords onto IP

View File

@ -41,17 +41,15 @@ type router struct {
func (r *router) init(core *Core) {
r.core = core
r.addr = *address_addrForNodeID(&r.core.dht.nodeID)
in := make(chan []byte, 32) // TODO something better than this...
in := make(chan []byte, 32) // TODO something better than this...
p := r.core.peers.newPeer(&r.core.boxPub, &r.core.sigPub) //, out, in)
p.out = func(packet []byte) {
// This is to make very sure it never blocks
for {
select {
case in <- packet:
return
default:
util_putBytes(<-in)
}
select {
case in <- packet:
return
default:
util_putBytes(packet)
}
}
r.in = in
@ -147,13 +145,10 @@ func (r *router) sendPacket(bs []byte) {
fallthrough
//default: go func() { sinfo.send<-bs }()
default:
for {
select {
case sinfo.send <- bs:
return
default:
util_putBytes(<-sinfo.send)
}
select {
case sinfo.send <- bs:
default:
util_putBytes(bs)
}
}
}
@ -191,7 +186,6 @@ func (r *router) handleIn(packet []byte) {
case wire_ProtocolTraffic:
r.handleProto(packet)
default: /*panic("Should not happen in testing") ;*/
return
}
}
@ -206,13 +200,10 @@ func (r *router) handleTraffic(packet []byte) {
return
}
//go func () { sinfo.recv<-&p }()
for {
select {
case sinfo.recv <- &p:
return
default:
util_putBytes((<-sinfo.recv).payload)
}
select {
case sinfo.recv <- &p:
default:
util_putBytes(p.payload)
}
}

View File

@ -176,13 +176,10 @@ func (iface *tcpInterface) handler(sock *net.TCPConn) {
}()
p.out = func(msg []byte) {
defer func() { recover() }()
for {
select {
case out <- msg:
return
default:
util_putBytes(<-out)
}
select {
case out <- msg:
default:
util_putBytes(msg)
}
}
sock.SetNoDelay(true)

View File

@ -224,13 +224,10 @@ func (iface *udpInterface) handleKeys(msg []byte, addr connAddr) {
}
conn.peer.out = func(msg []byte) {
defer func() { recover() }()
for {
select {
case conn.out <- msg:
return
default:
util_putBytes(<-conn.out)
}
select {
case conn.out <- msg:
default:
util_putBytes(msg)
}
}
go func() {