debug: add gc-client TODO rm before merge
add automatic add and cat for debugging purposes retry cats cast string argh! @whyrusleepgin use mock repo to avoid clashing larger files for more variety FIXUP into gc-client fix: change in both places FIXUP gc-client fix(gc-client) start with size 1 (unit) fix(gc-client) start at 1 FIXUP prev gc-client rm comment gc-client larger files fix(gc-client) cat the data out to dev null to ensure entire file is fetched perf(gc-client) use io.Pipe to handle io transfer fixup
Brian Tiger Chow committed
Feb 12, 2015 at 14:19 UTC
31587b2d0a91179fd5b10fb4be772f6b9ea8c8fa
2 files changed
+257
test/supernode_client/.gitignore
new
+1
@@ -0,0 +1 @@
1
+.go-ipfs/
test/supernode_client/main.go
new
+256
@@ -0,0 +1,256 @@
1
+package main
2
+
3
+import (
4
+ "bytes"
5
+ "flag"
6
+ "fmt"
7
+ "io"
8
+ "io/ioutil"
9
+ "log"
10
+ "math"
11
+ "os"
12
+ gopath "path"
13
+ "time"
14
+
15
+ context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
16
+ ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
17
+ random "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-random"
18
+ "github.com/jbenet/go-ipfs/util/ipfsaddr"
19
+
20
+ "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
21
+ syncds "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/sync"
22
+ commands "github.com/jbenet/go-ipfs/commands"
23
+ core "github.com/jbenet/go-ipfs/core"
24
+ corehttp "github.com/jbenet/go-ipfs/core/corehttp"
25
+ corerouting "github.com/jbenet/go-ipfs/core/corerouting"
26
+ "github.com/jbenet/go-ipfs/core/coreunix"
27
+ peer "github.com/jbenet/go-ipfs/p2p/peer"
28
+ "github.com/jbenet/go-ipfs/repo"
29
+ config "github.com/jbenet/go-ipfs/repo/config"
30
+ fsrepo "github.com/jbenet/go-ipfs/repo/fsrepo"
31
+ eventlog "github.com/jbenet/go-ipfs/thirdparty/eventlog"
32
+ unit "github.com/jbenet/go-ipfs/thirdparty/unit"
33
+ ds2 "github.com/jbenet/go-ipfs/util/datastore2"
34
+)
35
+
36
+var elog = eventlog.Logger("gc-client")
37
+
38
+var (
39
+ cat = flag.Bool("cat", false, "else add")
40
+ seed = flag.Int64("seed", 1, "")
41
+ nBitsForKeypair = flag.Int("b", 1024, "number of bits for keypair (if repo is uninitialized)")
42
+)
43
+
44
+var bootstrapAddresses = []string{
45
+ "/ip4/104.236.70.34/tcp/4001/ipfs/QmaWJw5mcWkCThPtC7hVq28e3WbwLHnWF8HbMNJrRDybE4",
46
+ "/ip4/128.199.72.111/tcp/4001/ipfs/Qmd2cSiZUt7vhiuTmqBB7XWbkuFR8KMLiEuQssjyNXyaZT",
47
+ "/ip4/162.243.251.152/tcp/4001/ipfs/QmeaDCMmFWDuF4baxhMuvQH8APoymtMSJwhyZqUp9ux3SN",
48
+ "/ip4/178.62.8.190/tcp/4001/ipfs/QmdXzZ25cyzSF99csCQmmPZ1NTbWTe8qtKFaZKpZQPdTFB",
49
+ "/ip4/188.166.8.195/tcp/4001/ipfs/QmNU1Vpryj5hfSmybSYHnS497ttgy9aNJ3T2B8wY2uMso4",
50
+ "/ip4/188.226.225.73/tcp/4001/ipfs/QmYHGLxLfHFd9veUt159LMhLD8LqhDjBUS9ZehLkPap6od",
51
+ "/ip4/192.241.209.121/tcp/4001/ipfs/QmNPM851fxWb3qjLV66a8Nqtw7fBnMw2UrdFQidSdBQnvS",
52
+}
53
+
54
+func main() {
55
+ flag.Parse()
56
+ if err := run(); err != nil {
57
+ fmt.Fprintf(os.Stderr, "error: %s\n", err)
58
+ os.Exit(1)
59
+ }
60
+}
61
+
62
+func run() error {
63
+ fmt.Println("using gcr remotes:")
64
+ for _, p := range bootstrapAddresses {
65
+ fmt.Println("\t", p)
66
+ }
67
+
68
+ ctx, cancel := context.WithCancel(context.Background())
69
+ defer cancel()
70
+
71
+ cwd, err := os.Getwd()
72
+ if err != nil {
73
+ return err
74
+ }
75
+ repoPath := gopath.Join(cwd, ".go-ipfs")
76
+ if err := ensureRepoInitialized(repoPath); err != nil {
77
+ }
78
+ repo := fsrepo.At(repoPath)
79
+ if err := repo.Open(); err != nil { // owned by node
80
+ return err
81
+ }
82
+ cfg := repo.Config()
83
+ cfg.Bootstrap = bootstrapAddresses
84
+ if err := repo.SetConfig(cfg); err != nil {
85
+ return err
86
+ }
87
+
88
+ var addrs []ipfsaddr.IPFSAddr
89
+ for _, info := range bootstrapAddresses {
90
+ addr, err := ipfsaddr.ParseString(info)
91
+ if err != nil {
92
+ return err
93
+ }
94
+ addrs = append(addrs, addr)
95
+ }
96
+
97
+ var infos []peer.PeerInfo
98
+ for _, addr := range addrs {
99
+ infos = append(infos, peer.PeerInfo{
100
+ ID: addr.ID(),
101
+ Addrs: []ma.Multiaddr{addr.Transport()},
102
+ })
103
+ }
104
+
105
+ node, err := core.NewIPFSNode(
106
+ ctx,
107
+ core.OnlineWithOptions(
108
+ repo,
109
+ corerouting.SupernodeClient(infos...),
110
+ core.DefaultHostOption,
111
+ ),
112
+ )
113
+ if err != nil {
114
+ return err
115
+ }
116
+ defer node.Close()
117
+
118
+ opts := []corehttp.ServeOption{
119
+ corehttp.CommandsOption(cmdCtx(node, repoPath)),
120
+ corehttp.GatewayOption(false),
121
+ }
122
+
123
+ if *cat {
124
+ if err := runFileCattingWorker(ctx, node); err != nil {
125
+ return err
126
+ }
127
+ } else {
128
+ if err := runFileAddingWorker(node); err != nil {
129
+ return err
130
+ }
131
+ }
132
+ return corehttp.ListenAndServe(node, cfg.Addresses.API, opts...)
133
+}
134
+
135
+func ensureRepoInitialized(path string) error {
136
+ if !fsrepo.IsInitialized(path) {
137
+ conf, err := config.Init(ioutil.Discard, *nBitsForKeypair)
138
+ if err != nil {
139
+ return err
140
+ }
141
+ if err := fsrepo.Init(path, conf); err != nil {
142
+ return err
143
+ }
144
+ }
145
+ return nil
146
+}
147
+
148
+func sizeOfIthFile(i int64) int64 {
149
+ return (1 << uint64(i)) * unit.KB
150
+}
151
+
152
+func runFileAddingWorker(n *core.IpfsNode) error {
153
+ go func() {
154
+ var i int64
155
+ for i = 1; i < math.MaxInt32; i++ {
156
+ piper, pipew := io.Pipe()
157
+ go func() {
158
+ defer pipew.Close()
159
+ if err := random.WritePseudoRandomBytes(sizeOfIthFile(i), pipew, *seed); err != nil {
160
+ log.Fatal(err)
161
+ }
162
+ }()
163
+ k, err := coreunix.Add(n, piper)
164
+ if err != nil {
165
+ log.Fatal(err)
166
+ }
167
+ log.Println("added file", "seed", *seed, "#", i, "key", k, "size", unit.Information(sizeOfIthFile(i)))
168
+ time.Sleep(1 * time.Second)
169
+ }
170
+ }()
171
+ return nil
172
+}
173
+
174
+func runFileCattingWorker(ctx context.Context, n *core.IpfsNode) error {
175
+ conf, err := config.Init(ioutil.Discard, *nBitsForKeypair)
176
+ if err != nil {
177
+ return err
178
+ }
179
+
180
+ dummy, err := core.NewIPFSNode(ctx, core.Offline(&repo.Mock{
181
+ D: ds2.CloserWrap(syncds.MutexWrap(datastore.NewMapDatastore())),
182
+ C: *conf,
183
+ }))
184
+ if err != nil {
185
+ return err
186
+ }
187
+
188
+ go func() {
189
+ defer dummy.Close()
190
+ var i int64 = 1
191
+ for {
192
+ var buf bytes.Buffer
193
+ if err := random.WritePseudoRandomBytes(sizeOfIthFile(i), &buf, *seed); err != nil {
194
+ log.Fatal(err)
195
+ }
196
+ // add to a dummy node to discover the key
197
+ k, err := coreunix.Add(dummy, bytes.NewReader(buf.Bytes()))
198
+ if err != nil {
199
+ log.Fatal(err)
200
+ }
201
+ e := elog.EventBegin(ctx, "cat", eventlog.LoggableF(func() map[string]interface{} {
202
+ return map[string]interface{}{
203
+ "key": k,
204
+ "localPeer": n.Identity,
205
+ }
206
+ }))
207
+ if r, err := coreunix.Cat(n, k); err != nil {
208
+ e.Done()
209
+ log.Printf("failed to cat file. seed: %d #%d key: %s err: %s", *seed, i, k, err)
210
+ } else {
211
+ log.Println("found file", "seed", *seed, "#", i, "key", k, "size", unit.Information(sizeOfIthFile(i)))
212
+ io.Copy(ioutil.Discard, r)
213
+ e.Done()
214
+ log.Println("catted file", "seed", *seed, "#", i, "key", k, "size", unit.Information(sizeOfIthFile(i)))
215
+ i++
216
+ }
217
+ time.Sleep(time.Second)
218
+ }
219
+ }()
220
+ return nil
221
+}
222
+
223
+func toPeerInfos(bpeers []config.BootstrapPeer) ([]peer.PeerInfo, error) {
224
+ var peers []peer.PeerInfo
225
+ for _, bootstrap := range bpeers {
226
+ p, err := toPeerInfo(bootstrap)
227
+ if err != nil {
228
+ return nil, err
229
+ }
230
+ peers = append(peers, p)
231
+ }
232
+ return peers, nil
233
+}
234
+
235
+func toPeerInfo(bootstrap config.BootstrapPeer) (p peer.PeerInfo, err error) {
236
+ p = peer.PeerInfo{
237
+ ID: bootstrap.ID(),
238
+ Addrs: []ma.Multiaddr{bootstrap.Multiaddr()},
239
+ }
240
+ return p, nil
241
+}
242
+
243
+func cmdCtx(node *core.IpfsNode, repoPath string) commands.Context {
244
+ return commands.Context{
245
+ // TODO deprecate this shit
246
+ Context: context.Background(),
247
+ Online: true,
248
+ ConfigRoot: repoPath,
249
+ LoadConfig: func(path string) (*config.Config, error) {
250
+ return node.Repo.Config(), nil
251
+ },
252
+ ConstructNode: func() (*core.IpfsNode, error) {
253
+ return node, nil
254
+ },
255
+ }
256
+}