Merge pull request #42 from ipfs/feat/bandwidth-limited-tests

Feat/bandwidth limited tests

Author: Stebalien

.gitignore

@@ -0,0 +1 @@
+tmp

dup_blocks_test.go renamed to benchmarks_test.go

@@ -9,9 +9,10 @@ import (
 	"testing"
 	"time"
 
-	tn "github.com/ipfs/go-bitswap/testnet"
+	"github.com/ipfs/go-bitswap/testutil"
 
 	bssession "github.com/ipfs/go-bitswap/session"
+	tn "github.com/ipfs/go-bitswap/testnet"
 	"github.com/ipfs/go-block-format"
 	cid "github.com/ipfs/go-cid"
 	blocksutil "github.com/ipfs/go-ipfs-blocksutil"
@@ -34,6 +35,7 @@ type runStats struct {
 var benchmarkLog []runStats
 
 func BenchmarkDups2Nodes(b *testing.B) {
+	benchmarkLog = nil
 	fixedDelay := delay.Fixed(10 * time.Millisecond)
 	b.Run("AllToAll-OneAtATime", func(b *testing.B) {
 		subtestDistributeAndFetch(b, 3, 100, fixedDelay, allToAll, oneAtATime)
@@ -93,50 +95,83 @@ func BenchmarkDups2Nodes(b *testing.B) {
 		subtestDistributeAndFetch(b, 200, 20, fixedDelay, allToAll, batchFetchAll)
 	})
 	out, _ := json.MarshalIndent(benchmarkLog, "", "  ")
-	ioutil.WriteFile("benchmark.json", out, 0666)
+	ioutil.WriteFile("tmp/benchmark.json", out, 0666)
 }
 
 const fastSpeed = 60 * time.Millisecond
 const mediumSpeed = 200 * time.Millisecond
 const slowSpeed = 800 * time.Millisecond
 const superSlowSpeed = 4000 * time.Millisecond
 const distribution = 20 * time.Millisecond
+const fastBandwidth = 1250000.0
+const fastBandwidthDeviation = 300000.0
+const mediumBandwidth = 500000.0
+const mediumBandwidthDeviation = 80000.0
+const slowBandwidth = 100000.0
+const slowBandwidthDeviation = 16500.0
+const stdBlockSize = 8000
 
 func BenchmarkDupsManyNodesRealWorldNetwork(b *testing.B) {
+	benchmarkLog = nil
 	fastNetworkDelayGenerator := tn.InternetLatencyDelayGenerator(
 		mediumSpeed-fastSpeed, slowSpeed-fastSpeed,
 		0.0, 0.0, distribution, nil)
 	fastNetworkDelay := delay.Delay(fastSpeed, fastNetworkDelayGenerator)
+	fastBandwidthGenerator := tn.VariableRateLimitGenerator(fastBandwidth, fastBandwidthDeviation, nil)
 	averageNetworkDelayGenerator := tn.InternetLatencyDelayGenerator(
 		mediumSpeed-fastSpeed, slowSpeed-fastSpeed,
 		0.3, 0.3, distribution, nil)
 	averageNetworkDelay := delay.Delay(fastSpeed, averageNetworkDelayGenerator)
+	averageBandwidthGenerator := tn.VariableRateLimitGenerator(mediumBandwidth, mediumBandwidthDeviation, nil)
 	slowNetworkDelayGenerator := tn.InternetLatencyDelayGenerator(
 		mediumSpeed-fastSpeed, superSlowSpeed-fastSpeed,
 		0.3, 0.3, distribution, nil)
 	slowNetworkDelay := delay.Delay(fastSpeed, slowNetworkDelayGenerator)
+	slowBandwidthGenerator := tn.VariableRateLimitGenerator(slowBandwidth, slowBandwidthDeviation, nil)
 
 	b.Run("200Nodes-AllToAll-BigBatch-FastNetwork", func(b *testing.B) {
-		subtestDistributeAndFetch(b, 300, 200, fastNetworkDelay, allToAll, batchFetchAll)
+		subtestDistributeAndFetchRateLimited(b, 300, 200, fastNetworkDelay, fastBandwidthGenerator, stdBlockSize, allToAll, batchFetchAll)
 	})
 	b.Run("200Nodes-AllToAll-BigBatch-AverageVariableSpeedNetwork", func(b *testing.B) {
-		subtestDistributeAndFetch(b, 300, 200, averageNetworkDelay, allToAll, batchFetchAll)
+		subtestDistributeAndFetchRateLimited(b, 300, 200, averageNetworkDelay, averageBandwidthGenerator, stdBlockSize, allToAll, batchFetchAll)
 	})
 	b.Run("200Nodes-AllToAll-BigBatch-SlowVariableSpeedNetwork", func(b *testing.B) {
-		subtestDistributeAndFetch(b, 300, 200, slowNetworkDelay, allToAll, batchFetchAll)
+		subtestDistributeAndFetchRateLimited(b, 300, 200, slowNetworkDelay, slowBandwidthGenerator, stdBlockSize, allToAll, batchFetchAll)
 	})
+	out, _ := json.MarshalIndent(benchmarkLog, "", "  ")
+	ioutil.WriteFile("tmp/rw-benchmark.json", out, 0666)
 }
 
 func subtestDistributeAndFetch(b *testing.B, numnodes, numblks int, d delay.D, df distFunc, ff fetchFunc) {
 	start := time.Now()
 	net := tn.VirtualNetwork(mockrouting.NewServer(), d)
+
 	sg := NewTestSessionGenerator(net)
 	defer sg.Close()
 
 	bg := blocksutil.NewBlockGenerator()
 
 	instances := sg.Instances(numnodes)
 	blocks := bg.Blocks(numblks)
+	runDistribution(b, instances, blocks, df, ff, start)
+}
+
+func subtestDistributeAndFetchRateLimited(b *testing.B, numnodes, numblks int, d delay.D, rateLimitGenerator tn.RateLimitGenerator, blockSize int64, df distFunc, ff fetchFunc) {
+	start := time.Now()
+	net := tn.RateLimitedVirtualNetwork(mockrouting.NewServer(), d, rateLimitGenerator)
+
+	sg := NewTestSessionGenerator(net)
+	defer sg.Close()
+
+	instances := sg.Instances(numnodes)
+	blocks := testutil.GenerateBlocksOfSize(numblks, blockSize)
+
+	runDistribution(b, instances, blocks, df, ff, start)
+}
+
+func runDistribution(b *testing.B, instances []Instance, blocks []blocks.Block, df distFunc, ff fetchFunc, start time.Time) {
+
+	numnodes := len(instances)
 
 	fetcher := instances[numnodes-1]
 

package.json

@@ -9,9 +9,9 @@
   "gxDependencies": [
     {
       "author": "whyrusleeping",
-      "hash": "QmdJdFQc5U3RAKgJQGmWR7SSM7TLuER5FWz5Wq6Tzs2CnS",
+      "hash": "QmYxivS34F2M2n44WQQnRHGAKS8aoRUxwGpi9wk4Cdn4Jf",
       "name": "go-libp2p",
-      "version": "6.0.29"
+      "version": "6.0.30"
     },
     {
       "author": "hsanjuan",
@@ -184,6 +184,12 @@
       "hash": "Qmf7HqcW7LtCi1W8y2bdx2eJpze74jkbKqpByxgXikdbLF",
       "name": "go-detect-race",
       "version": "1.0.1"
+    },
+    {
+      "author": "jbenet",
+      "hash": "QmSJ9n2s9NUoA9D849W5jj5SJ94nMcZpj1jCgQJieiNqSt",
+      "name": "go-random",
+      "version": "1.0.0"
     }
   ],
   "gxVersion": "0.12.1",

testnet/rate_limit_generators.go

@@ -0,0 +1,42 @@
+package bitswap
+
+import (
+	"math/rand"
+)
+
+type fixedRateLimitGenerator struct {
+	rateLimit float64
+}
+
+// FixedRateLimitGenerator returns a rate limit generatoe that always generates
+// the specified rate limit in bytes/sec.
+func FixedRateLimitGenerator(rateLimit float64) RateLimitGenerator {
+	return &fixedRateLimitGenerator{rateLimit}
+}
+
+func (rateLimitGenerator *fixedRateLimitGenerator) NextRateLimit() float64 {
+	return rateLimitGenerator.rateLimit
+}
+
+type variableRateLimitGenerator struct {
+	rateLimit float64
+	std       float64
+	rng       *rand.Rand
+}
+
+// VariableRateLimitGenerator makes rate limites that following a normal distribution.
+func VariableRateLimitGenerator(rateLimit float64, std float64, rng *rand.Rand) RateLimitGenerator {
+	if rng == nil {
+		rng = sharedRNG
+	}
+
+	return &variableRateLimitGenerator{
+		std:       std,
+		rng:       rng,
+		rateLimit: rateLimit,
+	}
+}
+
+func (rateLimitGenerator *variableRateLimitGenerator) NextRateLimit() float64 {
+	return rateLimitGenerator.rng.NormFloat64()*rateLimitGenerator.std + rateLimitGenerator.rateLimit
+}

testnet/virtual.go

@@ -18,28 +18,51 @@ import (
 	ifconnmgr "github.com/libp2p/go-libp2p-interface-connmgr"
 	peer "github.com/libp2p/go-libp2p-peer"
 	routing "github.com/libp2p/go-libp2p-routing"
+	mocknet "github.com/libp2p/go-libp2p/p2p/net/mock"
 	testutil "github.com/libp2p/go-testutil"
 )
 
 var log = logging.Logger("bstestnet")
 
 func VirtualNetwork(rs mockrouting.Server, d delay.D) Network {
 	return &network{
-		latencies:     make(map[peer.ID]map[peer.ID]time.Duration),
-		clients:       make(map[peer.ID]*receiverQueue),
-		delay:         d,
-		routingserver: rs,
-		conns:         make(map[string]struct{}),
+		latencies:          make(map[peer.ID]map[peer.ID]time.Duration),
+		clients:            make(map[peer.ID]*receiverQueue),
+		delay:              d,
+		routingserver:      rs,
+		isRateLimited:      false,
+		rateLimitGenerator: nil,
+		conns:              make(map[string]struct{}),
+	}
+}
+
+type RateLimitGenerator interface {
+	NextRateLimit() float64
+}
+
+func RateLimitedVirtualNetwork(rs mockrouting.Server, d delay.D, rateLimitGenerator RateLimitGenerator) Network {
+	return &network{
+		latencies:          make(map[peer.ID]map[peer.ID]time.Duration),
+		rateLimiters:       make(map[peer.ID]map[peer.ID]*mocknet.RateLimiter),
+		clients:            make(map[peer.ID]*receiverQueue),
+		delay:              d,
+		routingserver:      rs,
+		isRateLimited:      true,
+		rateLimitGenerator: rateLimitGenerator,
+		conns:              make(map[string]struct{}),
 	}
 }
 
 type network struct {
-	mu            sync.Mutex
-	latencies     map[peer.ID]map[peer.ID]time.Duration
-	clients       map[peer.ID]*receiverQueue
-	routingserver mockrouting.Server
-	delay         delay.D
-	conns         map[string]struct{}
+	mu                 sync.Mutex
+	latencies          map[peer.ID]map[peer.ID]time.Duration
+	rateLimiters       map[peer.ID]map[peer.ID]*mocknet.RateLimiter
+	clients            map[peer.ID]*receiverQueue
+	routingserver      mockrouting.Server
+	delay              delay.D
+	isRateLimited      bool
+	rateLimitGenerator RateLimitGenerator
+	conns              map[string]struct{}
 }
 
 type message struct {
@@ -102,6 +125,26 @@ func (n *network) SendMessage(
 		latencies[to] = latency
 	}
 
+	var bandwidthDelay time.Duration
+	if n.isRateLimited {
+		rateLimiters, ok := n.rateLimiters[from]
+		if !ok {
+			rateLimiters = make(map[peer.ID]*mocknet.RateLimiter)
+			n.rateLimiters[from] = rateLimiters
+		}
+
+		rateLimiter, ok := rateLimiters[to]
+		if !ok {
+			rateLimiter = mocknet.NewRateLimiter(n.rateLimitGenerator.NextRateLimit())
+			rateLimiters[to] = rateLimiter
+		}
+
+		size := mes.ToProtoV1().Size()
+		bandwidthDelay = rateLimiter.Limit(size)
+	} else {
+		bandwidthDelay = 0
+	}
+
 	receiver, ok := n.clients[to]
 	if !ok {
 		return errors.New("cannot locate peer on network")
@@ -113,7 +156,7 @@ func (n *network) SendMessage(
 	msg := &message{
 		from:       from,
 		msg:        mes,
-		shouldSend: time.Now().Add(latency),
+		shouldSend: time.Now().Add(latency).Add(bandwidthDelay),
 	}
 	receiver.enqueue(msg)
 

testutil/testutil.go

@@ -1,6 +1,10 @@
 package testutil
 
 import (
+	"bytes"
+
+	random "github.com/jbenet/go-random"
+
 	bsmsg "github.com/ipfs/go-bitswap/message"
 	"github.com/ipfs/go-bitswap/wantlist"
 	"github.com/ipfs/go-block-format"
@@ -11,6 +15,25 @@ import (
 
 var blockGenerator = blocksutil.NewBlockGenerator()
 var prioritySeq int
+var seedSeq int64
+
+func randomBytes(n int64, seed int64) []byte {
+	data := new(bytes.Buffer)
+	random.WritePseudoRandomBytes(n, data, seed)
+	return data.Bytes()
+}
+
+// GenerateBlocksOfSize generates a series of blocks of the given byte size
+func GenerateBlocksOfSize(n int, size int64) []blocks.Block {
+	generatedBlocks := make([]blocks.Block, 0, n)
+	for i := 0; i < n; i++ {
+		seedSeq++
+		b := blocks.NewBlock(randomBytes(size, seedSeq))
+		generatedBlocks = append(generatedBlocks, b)
+
+	}
+	return generatedBlocks
+}
 
 // GenerateCids produces n content identifiers.
 func GenerateCids(n int) []cid.Cid {