454 lines
12 KiB
Go
Raw Normal View History

2017-12-28 22:16:20 -06:00
package main
import "fmt"
import "bufio"
import "os"
import "strings"
import "strconv"
import "time"
import "log"
2017-12-28 22:16:20 -06:00
2018-11-25 12:25:38 -06:00
import "runtime"
2017-12-28 22:16:20 -06:00
import "runtime/pprof"
import "flag"
2018-12-07 19:56:04 -06:00
import . "github.com/yggdrasil-network/yggdrasil-go/src/yggdrasil"
2017-12-28 22:16:20 -06:00
////////////////////////////////////////////////////////////////////////////////
type Node struct {
2018-01-04 22:37:51 +00:00
index int
core Core
send chan<- []byte
recv <-chan []byte
2017-12-28 22:16:20 -06:00
}
func (n *Node) init(index int) {
2018-01-04 22:37:51 +00:00
n.index = index
n.core.Init()
n.send = n.core.DEBUG_getSend()
n.recv = n.core.DEBUG_getRecv()
n.core.DEBUG_simFixMTU()
2017-12-28 22:16:20 -06:00
}
func (n *Node) printTraffic() {
2018-01-04 22:37:51 +00:00
for {
packet := <-n.recv
fmt.Println(n.index, packet)
//panic("Got a packet")
}
2017-12-28 22:16:20 -06:00
}
func (n *Node) startPeers() {
2018-01-04 22:37:51 +00:00
//for _, p := range n.core.Peers.Ports {
// go p.MainLoop()
//}
//go n.printTraffic()
//n.core.Peers.DEBUG_startPeers()
2017-12-28 22:16:20 -06:00
}
func linkNodes(m, n *Node) {
2018-01-04 22:37:51 +00:00
// Don't allow duplicates
if m.core.DEBUG_getPeers().DEBUG_hasPeer(n.core.DEBUG_getSigningPublicKey()) {
2018-01-04 22:37:51 +00:00
return
}
// Create peers
// Buffering reduces packet loss in the sim
// This slightly speeds up testing (fewer delays before retrying a ping)
pLinkPub, pLinkPriv := m.core.DEBUG_newBoxKeys()
qLinkPub, qLinkPriv := m.core.DEBUG_newBoxKeys()
p := m.core.DEBUG_getPeers().DEBUG_newPeer(n.core.DEBUG_getEncryptionPublicKey(),
n.core.DEBUG_getSigningPublicKey(), *m.core.DEBUG_getSharedKey(pLinkPriv, qLinkPub))
q := n.core.DEBUG_getPeers().DEBUG_newPeer(m.core.DEBUG_getEncryptionPublicKey(),
m.core.DEBUG_getSigningPublicKey(), *n.core.DEBUG_getSharedKey(qLinkPriv, pLinkPub))
2018-01-04 22:37:51 +00:00
DEBUG_simLinkPeers(p, q)
return
2017-12-28 22:16:20 -06:00
}
func makeStoreSquareGrid(sideLength int) map[int]*Node {
2018-01-04 22:37:51 +00:00
store := make(map[int]*Node)
nNodes := sideLength * sideLength
idxs := make([]int, 0, nNodes)
// TODO shuffle nodeIDs
for idx := 1; idx <= nNodes; idx++ {
idxs = append(idxs, idx)
}
for _, idx := range idxs {
node := &Node{}
node.init(idx)
store[idx] = node
}
for idx := 0; idx < nNodes; idx++ {
if (idx % sideLength) != 0 {
linkNodes(store[idxs[idx]], store[idxs[idx-1]])
}
if idx >= sideLength {
linkNodes(store[idxs[idx]], store[idxs[idx-sideLength]])
}
}
//for _, node := range store { node.initPorts() }
return store
2017-12-28 22:16:20 -06:00
}
func makeStoreStar(nNodes int) map[int]*Node {
2018-01-04 22:37:51 +00:00
store := make(map[int]*Node)
center := &Node{}
center.init(0)
store[0] = center
for idx := 1; idx < nNodes; idx++ {
node := &Node{}
node.init(idx)
store[idx] = node
linkNodes(center, node)
}
return store
2017-12-28 22:16:20 -06:00
}
func loadGraph(path string) map[int]*Node {
2018-01-04 22:37:51 +00:00
f, err := os.Open(path)
if err != nil {
panic(err)
}
defer f.Close()
store := make(map[int]*Node)
s := bufio.NewScanner(f)
for s.Scan() {
line := s.Text()
nodeIdxstrs := strings.Split(line, " ")
nodeIdx0, _ := strconv.Atoi(nodeIdxstrs[0])
nodeIdx1, _ := strconv.Atoi(nodeIdxstrs[1])
if store[nodeIdx0] == nil {
node := &Node{}
node.init(nodeIdx0)
store[nodeIdx0] = node
}
if store[nodeIdx1] == nil {
node := &Node{}
node.init(nodeIdx1)
store[nodeIdx1] = node
}
linkNodes(store[nodeIdx0], store[nodeIdx1])
}
//for _, node := range store { node.initPorts() }
return store
2017-12-28 22:16:20 -06:00
}
////////////////////////////////////////////////////////////////////////////////
func startNetwork(store map[[32]byte]*Node) {
2018-01-04 22:37:51 +00:00
for _, node := range store {
node.startPeers()
}
2017-12-28 22:16:20 -06:00
}
func getKeyedStore(store map[int]*Node) map[[32]byte]*Node {
2018-01-04 22:37:51 +00:00
newStore := make(map[[32]byte]*Node)
for _, node := range store {
newStore[node.core.DEBUG_getSigningPublicKey()] = node
2018-01-04 22:37:51 +00:00
}
return newStore
2017-12-28 22:16:20 -06:00
}
func testPaths(store map[[32]byte]*Node) bool {
2018-01-04 22:37:51 +00:00
nNodes := len(store)
count := 0
for _, source := range store {
count++
fmt.Printf("Testing paths from node %d / %d (%d)\n", count, nNodes, source.index)
for _, dest := range store {
//if source == dest { continue }
destLoc := dest.core.DEBUG_getLocator()
coords := destLoc.DEBUG_getCoords()
temp := 0
ttl := ^uint64(0)
oldTTL := ttl
for here := source; here != dest; {
temp++
if temp > 4096 {
fmt.Println("Loop?")
time.Sleep(time.Second)
return false
2018-01-04 22:37:51 +00:00
}
2018-06-07 20:29:22 -05:00
nextPort := here.core.DEBUG_switchLookup(coords)
2018-01-04 22:37:51 +00:00
// First check if "here" is accepting packets from the previous node
// TODO explain how this works
ports := here.core.DEBUG_getPeers().DEBUG_getPorts()
nextPeer := ports[nextPort]
if nextPeer == nil {
fmt.Println("Peer associated with next port is nil")
return false
}
next := store[nextPeer.DEBUG_getSigKey()]
/*
if next == here {
//for idx, link := range here.links {
// fmt.Println("DUMP:", idx, link.nodeID)
//}
if nextPort != 0 { panic("This should not be") }
fmt.Println("Failed to route:", source.index, here.index, dest.index, oldTTL, ttl)
//here.table.DEBUG_dumpTable()
//fmt.Println("Ports:", here.nodeID, here.ports)
return false
panic(fmt.Sprintln("Routing Loop:",
source.index,
here.index,
dest.index))
}
*/
if temp > 4090 {
fmt.Println("DEBUG:",
source.index, source.core.DEBUG_getLocator(),
here.index, here.core.DEBUG_getLocator(),
dest.index, dest.core.DEBUG_getLocator())
//here.core.DEBUG_getSwitchTable().DEBUG_dumpTable()
2018-01-04 22:37:51 +00:00
}
if here != source {
// This is sufficient to check for routing loops or blackholes
//break
}
2018-06-07 20:18:13 -05:00
if here == next {
2018-06-07 20:29:22 -05:00
fmt.Println("Drop:", source.index, here.index, dest.index, oldTTL)
2018-06-07 20:18:13 -05:00
return false
}
2018-01-04 22:37:51 +00:00
here = next
}
}
}
return true
2017-12-28 22:16:20 -06:00
}
func stressTest(store map[[32]byte]*Node) {
2018-01-04 22:37:51 +00:00
fmt.Println("Stress testing network...")
nNodes := len(store)
dests := make([][]byte, 0, nNodes)
for _, dest := range store {
loc := dest.core.DEBUG_getLocator()
coords := loc.DEBUG_getCoords()
dests = append(dests, coords)
}
lookups := 0
start := time.Now()
for _, source := range store {
for _, coords := range dests {
2018-06-07 20:29:22 -05:00
source.core.DEBUG_switchLookup(coords)
2018-01-04 22:37:51 +00:00
lookups++
}
}
timed := time.Since(start)
fmt.Printf("%d lookups in %s (%f lookups per second)\n",
lookups,
timed,
float64(lookups)/timed.Seconds())
2017-12-28 22:16:20 -06:00
}
func pingNodes(store map[[32]byte]*Node) {
2018-01-04 22:37:51 +00:00
fmt.Println("Sending pings...")
nNodes := len(store)
count := 0
equiv := func(a []byte, b []byte) bool {
if len(a) != len(b) {
return false
}
for idx := 0; idx < len(a); idx++ {
if a[idx] != b[idx] {
return false
}
}
return true
}
for _, source := range store {
count++
//if count > 16 { break }
fmt.Printf("Sending packets from node %d/%d (%d)\n", count, nNodes, source.index)
sourceKey := source.core.DEBUG_getEncryptionPublicKey()
2018-01-04 22:37:51 +00:00
payload := sourceKey[:]
sourceAddr := source.core.DEBUG_getAddr()[:]
sendTo := func(bs []byte, destAddr []byte) {
packet := make([]byte, 40+len(bs))
copy(packet[8:24], sourceAddr)
copy(packet[24:40], destAddr)
copy(packet[40:], bs)
packet[0] = 6 << 4
2018-01-04 22:37:51 +00:00
source.send <- packet
}
destCount := 0
for _, dest := range store {
destCount += 1
fmt.Printf("%d Nodes, %d Send, %d Recv\n", nNodes, count, destCount)
if dest == source {
fmt.Println("Skipping self")
continue
}
destAddr := dest.core.DEBUG_getAddr()[:]
ticker := time.NewTicker(150 * time.Millisecond)
2018-11-25 12:25:38 -06:00
sendTo(payload, destAddr)
2018-01-04 22:37:51 +00:00
for loop := true; loop; {
select {
case packet := <-dest.recv:
{
if equiv(payload, packet[len(packet)-len(payload):]) {
loop = false
}
}
2018-11-25 12:25:38 -06:00
case <-ticker.C:
2018-01-04 22:37:51 +00:00
sendTo(payload, destAddr)
//dumpDHTSize(store) // note that this uses racey functions to read things...
2018-01-04 22:37:51 +00:00
}
}
ticker.Stop()
}
//break // Only try sending pings from 1 node
// This is because, for some reason, stopTun() doesn't always close it
// And if two tuns are up, bad things happen (sends via wrong interface)
}
fmt.Println("Finished pinging nodes")
2017-12-28 22:16:20 -06:00
}
func pingBench(store map[[32]byte]*Node) {
2018-01-04 22:37:51 +00:00
fmt.Println("Benchmarking pings...")
nPings := 0
payload := make([]byte, 1280+40) // MTU + ipv6 header
var timed time.Duration
//nNodes := len(store)
count := 0
for _, source := range store {
count++
//fmt.Printf("Sending packets from node %d/%d (%d)\n", count, nNodes, source.index)
getPing := func(key [32]byte, decodedCoords []byte) []byte {
// TODO write some function to do this the right way, put... somewhere...
coords := DEBUG_wire_encode_coords(decodedCoords)
packet := make([]byte, 0, len(key)+len(coords)+len(payload))
packet = append(packet, key[:]...)
packet = append(packet, coords...)
packet = append(packet, payload[:]...)
return packet
}
for _, dest := range store {
key := dest.core.DEBUG_getEncryptionPublicKey()
2018-01-04 22:37:51 +00:00
loc := dest.core.DEBUG_getLocator()
coords := loc.DEBUG_getCoords()
ping := getPing(key, coords)
// TODO make sure the session is open first
start := time.Now()
for i := 0; i < 1000000; i++ {
source.send <- ping
nPings++
}
timed += time.Since(start)
break
}
break
}
fmt.Printf("Sent %d pings in %s (%f per second)\n",
nPings,
timed,
float64(nPings)/timed.Seconds())
2017-12-28 22:16:20 -06:00
}
func dumpStore(store map[NodeID]*Node) {
2018-01-04 22:37:51 +00:00
for _, node := range store {
fmt.Println("DUMPSTORE:", node.index, node.core.DEBUG_getLocator())
node.core.DEBUG_getSwitchTable().DEBUG_dumpTable()
}
2017-12-28 22:16:20 -06:00
}
func dumpDHTSize(store map[[32]byte]*Node) {
2018-01-04 22:37:51 +00:00
var min, max, sum int
for _, node := range store {
num := node.core.DEBUG_getDHTSize()
min = num
max = num
break
}
for _, node := range store {
num := node.core.DEBUG_getDHTSize()
if num < min {
min = num
}
if num > max {
max = num
}
sum += num
}
avg := float64(sum) / float64(len(store))
fmt.Printf("DHT min %d / avg %f / max %d\n", min, avg, max)
2017-12-28 22:16:20 -06:00
}
2018-06-06 16:49:23 -05:00
func (n *Node) startTCP(listen string) {
n.core.DEBUG_setupAndStartGlobalTCPInterface(listen)
}
2018-06-06 16:49:23 -05:00
func (n *Node) connectTCP(remoteAddr string) {
n.core.AddPeer(remoteAddr, remoteAddr)
}
2017-12-28 22:16:20 -06:00
////////////////////////////////////////////////////////////////////////////////
var cpuprofile = flag.String("cpuprofile", "", "write cpu profile `file`")
var memprofile = flag.String("memprofile", "", "write memory profile to this file")
func main() {
2018-01-04 22:37:51 +00:00
flag.Parse()
if *cpuprofile != "" {
f, err := os.Create(*cpuprofile)
if err != nil {
panic(fmt.Sprintf("could not create CPU profile: ", err))
}
if err := pprof.StartCPUProfile(f); err != nil {
panic(fmt.Sprintf("could not start CPU profile: ", err))
}
defer pprof.StopCPUProfile()
}
if *memprofile != "" {
f, err := os.Create(*memprofile)
if err != nil {
panic(fmt.Sprintf("could not create memory profile: ", err))
}
defer func() { pprof.WriteHeapProfile(f); f.Close() }()
}
fmt.Println("Test")
Util_testAddrIDMask()
2018-10-20 17:32:54 -05:00
idxstore := makeStoreSquareGrid(4)
2018-01-04 22:37:51 +00:00
//idxstore := makeStoreStar(256)
//idxstore := loadGraph("misc/sim/hype-2016-09-19.list")
2018-10-20 17:32:54 -05:00
//idxstore := loadGraph("misc/sim/fc00-2017-08-12.txt")
2018-01-04 22:37:51 +00:00
//idxstore := loadGraph("skitter")
kstore := getKeyedStore(idxstore)
//*
logger := log.New(os.Stderr, "", log.Flags())
for _, n := range kstore {
n.core.DEBUG_setLogger(logger)
}
//*/
2018-01-04 22:37:51 +00:00
startNetwork(kstore)
//time.Sleep(10*time.Second)
// Note that testPaths only works if pressure is turend off
// Otherwise congestion can lead to routing loops?
for finished := false; !finished; {
finished = testPaths(kstore)
}
pingNodes(kstore)
//pingBench(kstore) // Only after disabling debug output
//stressTest(kstore)
//time.Sleep(120 * time.Second)
dumpDHTSize(kstore) // note that this uses racey functions to read things...
if false {
// This connects the sim to the local network
for _, node := range kstore {
2018-06-06 16:49:23 -05:00
node.startTCP("localhost:0")
node.connectTCP("localhost:12345")
break // just 1
}
for _, node := range kstore {
go func() {
// Just dump any packets sent to this node
for range node.recv {
}
}()
}
var block chan struct{}
<-block
}
2018-11-25 12:25:38 -06:00
runtime.GC()
2017-12-28 22:16:20 -06:00
}