Skip to content

Commit bd2d62f

Browse files
hannahhowardhannahhoward
authored
Benchmark framework + First memory fixes (#89)
* feat(benchmarks): initial benchmark infrastructure * fix(cidlink): mem allocations around link loading * fix(deps): update to latest deps use latest go-ipld-prime & go-ipld-prime-proto fixes * fix(deps): remove unused badger code
1 parent e98fd78 commit bd2d62f

17 files changed

+1055
-39
lines changed

‎benchmarks/benchmark_test.go

+207
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,207 @@
1+
package graphsync_test
2+
3+
import (
4+
"bytes"
5+
"context"
6+
"crypto/rand"
7+
"fmt"
8+
"io/ioutil"
9+
"os"
10+
"runtime"
11+
"strings"
12+
"sync"
13+
"sync/atomic"
14+
"testing"
15+
"time"
16+
17+
"github.com/ipfs/go-blockservice"
18+
"github.com/ipfs/go-cid"
19+
"github.com/ipfs/go-graphsync/benchmarks/testinstance"
20+
tn "github.com/ipfs/go-graphsync/benchmarks/testnet"
21+
blockstore "github.com/ipfs/go-ipfs-blockstore"
22+
chunker "github.com/ipfs/go-ipfs-chunker"
23+
delay "github.com/ipfs/go-ipfs-delay"
24+
offline "github.com/ipfs/go-ipfs-exchange-offline"
25+
files "github.com/ipfs/go-ipfs-files"
26+
ipldformat "github.com/ipfs/go-ipld-format"
27+
"github.com/ipfs/go-merkledag"
28+
"github.com/ipfs/go-unixfs/importer/balanced"
29+
ihelper "github.com/ipfs/go-unixfs/importer/helpers"
30+
cidlink "github.com/ipld/go-ipld-prime/linking/cid"
31+
basicnode "github.com/ipld/go-ipld-prime/node/basic"
32+
ipldselector "github.com/ipld/go-ipld-prime/traversal/selector"
33+
"github.com/ipld/go-ipld-prime/traversal/selector/builder"
34+
"github.com/stretchr/testify/require"
35+
)
36+
37+
const stdBlockSize = 8000
38+
39+
type runStats struct {
40+
Time time.Duration
41+
Name string
42+
}
43+
44+
var benchmarkLog []runStats
45+
46+
func BenchmarkRoundtripSuccess(b *testing.B) {
47+
ctx := context.Background()
48+
tdm, err := newTempDirMaker(b)
49+
require.NoError(b, err)
50+
b.Run("test-20-10000", func(b *testing.B) {
51+
subtestDistributeAndFetch(ctx, b, 20, delay.Fixed(0), time.Duration(0), allFilesUniformSize(10000), tdm)
52+
})
53+
}
54+
55+
func subtestDistributeAndFetch(ctx context.Context, b *testing.B, numnodes int, d delay.D, bstoreLatency time.Duration, df distFunc, tdm *tempDirMaker) {
56+
ctx, cancel := context.WithCancel(ctx)
57+
defer cancel()
58+
net := tn.VirtualNetwork(d)
59+
ig := testinstance.NewTestInstanceGenerator(ctx, net, nil, tdm)
60+
instances, err := ig.Instances(numnodes + b.N)
61+
require.NoError(b, err)
62+
destCids := df(ctx, b, instances[:numnodes])
63+
// Set the blockstore latency on seed nodes
64+
if bstoreLatency > 0 {
65+
for _, i := range instances {
66+
i.SetBlockstoreLatency(bstoreLatency)
67+
}
68+
}
69+
ssb := builder.NewSelectorSpecBuilder(basicnode.Style.Any)
70+
71+
allSelector := ssb.ExploreRecursive(ipldselector.RecursionLimitNone(),
72+
ssb.ExploreAll(ssb.ExploreRecursiveEdge())).Node()
73+
74+
runtime.GC()
75+
b.ResetTimer()
76+
b.ReportAllocs()
77+
for i := 0; i < b.N; i++ {
78+
fetcher := instances[i+numnodes]
79+
var wg sync.WaitGroup
80+
ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
81+
require.NoError(b, err)
82+
start := time.Now()
83+
for j := 0; j < numnodes; j++ {
84+
instance := instances[j]
85+
_, errChan := fetcher.Exchange.Request(ctx, instance.Peer, cidlink.Link{Cid: destCids[j]}, allSelector)
86+
87+
wg.Add(1)
88+
go func() {
89+
defer wg.Done()
90+
for {
91+
select {
92+
case <-ctx.Done():
93+
return
94+
case err, ok := <-errChan:
95+
if !ok {
96+
return
97+
}
98+
b.Fatalf("received error on request: %s", err.Error())
99+
}
100+
}
101+
}()
102+
}
103+
wg.Wait()
104+
result := runStats{
105+
Time: time.Since(start),
106+
Name: b.Name(),
107+
}
108+
benchmarkLog = append(benchmarkLog, result)
109+
cancel()
110+
fetcher.Close()
111+
}
112+
testinstance.Close(instances)
113+
ig.Close()
114+
115+
}
116+
117+
type distFunc func(ctx context.Context, b *testing.B, provs []testinstance.Instance) []cid.Cid
118+
119+
const unixfsChunkSize uint64 = 1 << 10
120+
const unixfsLinksPerLevel = 1024
121+
122+
func loadRandomUnixFxFile(ctx context.Context, b *testing.B, bs blockstore.Blockstore, size uint64) cid.Cid {
123+
124+
data := make([]byte, size)
125+
_, err := rand.Read(data)
126+
require.NoError(b, err)
127+
buf := bytes.NewReader(data)
128+
file := files.NewReaderFile(buf)
129+
130+
dagService := merkledag.NewDAGService(blockservice.New(bs, offline.Exchange(bs)))
131+
132+
// import to UnixFS
133+
bufferedDS := ipldformat.NewBufferedDAG(ctx, dagService)
134+
135+
params := ihelper.DagBuilderParams{
136+
Maxlinks: unixfsLinksPerLevel,
137+
RawLeaves: true,
138+
CidBuilder: nil,
139+
Dagserv: bufferedDS,
140+
}
141+
142+
db, err := params.New(chunker.NewSizeSplitter(file, int64(unixfsChunkSize)))
143+
require.NoError(b, err, "unable to setup dag builder")
144+
145+
nd, err := balanced.Layout(db)
146+
require.NoError(b, err, "unable to create unix fs node")
147+
148+
err = bufferedDS.Commit()
149+
require.NoError(b, err, "unable to commit unix fs node")
150+
151+
return nd.Cid()
152+
}
153+
154+
func allFilesUniformSize(size uint64) distFunc {
155+
return func(ctx context.Context, b *testing.B, provs []testinstance.Instance) []cid.Cid {
156+
cids := make([]cid.Cid, 0, len(provs))
157+
for _, prov := range provs {
158+
c := loadRandomUnixFxFile(ctx, b, prov.BlockStore, size)
159+
cids = append(cids, c)
160+
}
161+
return cids
162+
}
163+
}
164+
165+
type tempDirMaker struct {
166+
tdm string
167+
tempDirSeq int32
168+
b *testing.B
169+
}
170+
171+
var tempDirReplacer struct {
172+
sync.Once
173+
r *strings.Replacer
174+
}
175+
176+
// Cribbed from https://github.com/golang/go/blob/master/src/testing/testing.go#L890
177+
// and modified as needed due to https://github.com/golang/go/issues/41062
178+
func newTempDirMaker(b *testing.B) (*tempDirMaker, error) {
179+
c := &tempDirMaker{}
180+
// ioutil.TempDir doesn't like path separators in its pattern,
181+
// so mangle the name to accommodate subtests.
182+
tempDirReplacer.Do(func() {
183+
tempDirReplacer.r = strings.NewReplacer("/", "_", "\\", "_", ":", "_")
184+
})
185+
pattern := tempDirReplacer.r.Replace(b.Name())
186+
187+
var err error
188+
c.tdm, err = ioutil.TempDir("", pattern)
189+
if err != nil {
190+
return nil, err
191+
}
192+
b.Cleanup(func() {
193+
if err := os.RemoveAll(c.tdm); err != nil {
194+
b.Errorf("TempDir RemoveAll cleanup: %v", err)
195+
}
196+
})
197+
return c, nil
198+
}
199+
200+
func (tdm *tempDirMaker) TempDir() string {
201+
seq := atomic.AddInt32(&tdm.tempDirSeq, 1)
202+
dir := fmt.Sprintf("%s%c%03d", tdm.tdm, os.PathSeparator, seq)
203+
if err := os.Mkdir(dir, 0777); err != nil {
204+
tdm.b.Fatalf("TempDir: %v", err)
205+
}
206+
return dir
207+
}

‎benchmarks/testinstance/testinstance.go

+169
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,169 @@
1+
package testinstance
2+
3+
import (
4+
"context"
5+
"time"
6+
7+
"github.com/ipfs/go-datastore"
8+
ds "github.com/ipfs/go-datastore"
9+
"github.com/ipfs/go-datastore/delayed"
10+
ds_sync "github.com/ipfs/go-datastore/sync"
11+
graphsync "github.com/ipfs/go-graphsync"
12+
tn "github.com/ipfs/go-graphsync/benchmarks/testnet"
13+
gsimpl "github.com/ipfs/go-graphsync/impl"
14+
gsnet "github.com/ipfs/go-graphsync/network"
15+
"github.com/ipfs/go-graphsync/storeutil"
16+
blockstore "github.com/ipfs/go-ipfs-blockstore"
17+
delay "github.com/ipfs/go-ipfs-delay"
18+
"github.com/ipld/go-ipld-prime"
19+
peer "github.com/libp2p/go-libp2p-core/peer"
20+
p2ptestutil "github.com/libp2p/go-libp2p-netutil"
21+
tnet "github.com/libp2p/go-libp2p-testing/net"
22+
)
23+
24+
// TempDirGenerator is any interface that can generate temporary directories
25+
type TempDirGenerator interface {
26+
TempDir() string
27+
}
28+
29+
// NewTestInstanceGenerator generates a new InstanceGenerator for the given
30+
// testnet
31+
func NewTestInstanceGenerator(ctx context.Context, net tn.Network, gsOptions []gsimpl.Option, tempDirGenerator TempDirGenerator) InstanceGenerator {
32+
ctx, cancel := context.WithCancel(ctx)
33+
return InstanceGenerator{
34+
net: net,
35+
seq: 0,
36+
ctx: ctx, // TODO take ctx as param to Next, Instances
37+
cancel: cancel,
38+
gsOptions: gsOptions,
39+
tempDirGenerator: tempDirGenerator,
40+
}
41+
}
42+
43+
// InstanceGenerator generates new test instances of bitswap+dependencies
44+
type InstanceGenerator struct {
45+
seq int
46+
net tn.Network
47+
ctx context.Context
48+
cancel context.CancelFunc
49+
gsOptions []gsimpl.Option
50+
tempDirGenerator TempDirGenerator
51+
}
52+
53+
// Close closes the clobal context, shutting down all test instances
54+
func (g *InstanceGenerator) Close() error {
55+
g.cancel()
56+
return nil // for Closer interface
57+
}
58+
59+
// Next generates a new instance of graphsync + dependencies
60+
func (g *InstanceGenerator) Next() (Instance, error) {
61+
g.seq++
62+
p, err := p2ptestutil.RandTestBogusIdentity()
63+
if err != nil {
64+
return Instance{}, err
65+
}
66+
return NewInstance(g.ctx, g.net, p, g.gsOptions, g.tempDirGenerator.TempDir())
67+
}
68+
69+
// Instances creates N test instances of bitswap + dependencies and connects
70+
// them to each other
71+
func (g *InstanceGenerator) Instances(n int) ([]Instance, error) {
72+
var instances []Instance
73+
for j := 0; j < n; j++ {
74+
inst, err := g.Next()
75+
if err != nil {
76+
return nil, err
77+
}
78+
instances = append(instances, inst)
79+
}
80+
ConnectInstances(instances)
81+
return instances, nil
82+
}
83+
84+
// ConnectInstances connects the given instances to each other
85+
func ConnectInstances(instances []Instance) {
86+
for i, inst := range instances {
87+
for j := i + 1; j < len(instances); j++ {
88+
oinst := instances[j]
89+
err := inst.Adapter.ConnectTo(context.Background(), oinst.Peer)
90+
if err != nil {
91+
panic(err.Error())
92+
}
93+
}
94+
}
95+
}
96+
97+
// Close closes multiple instances at once
98+
func Close(instances []Instance) error {
99+
for _, i := range instances {
100+
if err := i.Close(); err != nil {
101+
return err
102+
}
103+
}
104+
return nil
105+
}
106+
107+
// Instance is a test instance of bitswap + dependencies for integration testing
108+
type Instance struct {
109+
Peer peer.ID
110+
Loader ipld.Loader
111+
Storer ipld.Storer
112+
Exchange graphsync.GraphExchange
113+
BlockStore blockstore.Blockstore
114+
Adapter gsnet.GraphSyncNetwork
115+
blockstoreDelay delay.D
116+
ds datastore.Batching
117+
}
118+
119+
// Close closes the associated datastore
120+
func (i *Instance) Close() error {
121+
return i.ds.Close()
122+
}
123+
124+
// Blockstore returns the block store for this test instance
125+
func (i *Instance) Blockstore() blockstore.Blockstore {
126+
return i.BlockStore
127+
}
128+
129+
// SetBlockstoreLatency customizes the artificial delay on receiving blocks
130+
// from a blockstore test instance.
131+
func (i *Instance) SetBlockstoreLatency(t time.Duration) time.Duration {
132+
return i.blockstoreDelay.Set(t)
133+
}
134+
135+
// NewInstance creates a test bitswap instance.
136+
//
137+
// NB: It's easy make mistakes by providing the same peer ID to two different
138+
// instances. To safeguard, use the InstanceGenerator to generate instances. It's
139+
// just a much better idea.
140+
func NewInstance(ctx context.Context, net tn.Network, p tnet.Identity, gsOptions []gsimpl.Option, tempDir string) (Instance, error) {
141+
bsdelay := delay.Fixed(0)
142+
143+
adapter := net.Adapter(p)
144+
dstore := ds_sync.MutexWrap(delayed.New(ds.NewMapDatastore(), bsdelay))
145+
bstore, err := blockstore.CachedBlockstore(ctx,
146+
blockstore.NewBlockstore(dstore),
147+
blockstore.DefaultCacheOpts())
148+
if err != nil {
149+
return Instance{}, err
150+
}
151+
152+
loader := storeutil.LoaderForBlockstore(bstore)
153+
storer := storeutil.StorerForBlockstore(bstore)
154+
gs := gsimpl.New(ctx, adapter, loader, storer, gsOptions...)
155+
gs.RegisterIncomingRequestHook(func(p peer.ID, request graphsync.RequestData, hookActions graphsync.IncomingRequestHookActions) {
156+
hookActions.ValidateRequest()
157+
})
158+
159+
return Instance{
160+
Adapter: adapter,
161+
Peer: p.ID(),
162+
Exchange: gs,
163+
Loader: loader,
164+
Storer: storer,
165+
BlockStore: bstore,
166+
blockstoreDelay: bsdelay,
167+
ds: dstore,
168+
}, nil
169+
}

‎benchmarks/testnet/interface.go

+16
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
1+
package testnet
2+
3+
import (
4+
gsnet "github.com/ipfs/go-graphsync/network"
5+
6+
"github.com/libp2p/go-libp2p-core/peer"
7+
tnet "github.com/libp2p/go-libp2p-testing/net"
8+
)
9+
10+
// Network is an interface for generating graphsync network interfaces
11+
// based on a test network.
12+
type Network interface {
13+
Adapter(tnet.Identity) gsnet.GraphSyncNetwork
14+
HasPeer(peer.ID) bool
15+
}
16+

0 commit comments

Comments
 (0)