From 90c8e05ee0512da0683d063c58e29cec90c5c09b Mon Sep 17 00:00:00 2001 From: lash Date: Sat, 16 Dec 2017 00:33:08 +0100 Subject: [PATCH] swarm/network, swarm/pss: Add context cancels, missing format vars --- swarm/network/protocol.go | 5 +-- .../simulations/discovery/discovery_test.go | 2 +- swarm/pss/client/client_test.go | 10 +++-- swarm/pss/handshake.go | 5 ++- swarm/pss/protocol.go | 2 +- swarm/pss/protocol_test.go | 6 ++- swarm/pss/pss.go | 2 +- swarm/pss/pss_test.go | 44 ++++++++++++------- 8 files changed, 46 insertions(+), 30 deletions(-) diff --git a/swarm/network/protocol.go b/swarm/network/protocol.go index 79471c8ec6..22c8334030 100644 --- a/swarm/network/protocol.go +++ b/swarm/network/protocol.go @@ -207,9 +207,8 @@ func (b *Bzz) RunProtocol(spec *protocols.Spec, run func(*bzzPeer) error) func(* // performHandshake implements the negotiation of the bzz handshake // shared among swarm subprotocols func performHandshake(p *protocols.Peer, handshake *HandshakeMsg) error { - ctx, _ := context.WithTimeout(context.Background(), bzzHandshakeTimeout) - // defer cancel() - // ctx, cancel := context.WithTimeout(context.Background(), bzzHandshakeTimeout) + ctx, cancel := context.WithTimeout(context.Background(), bzzHandshakeTimeout) + defer cancel() defer close(handshake.done) rsh, err := p.Handshake(ctx, handshake, checkHandshake) if err != nil { diff --git a/swarm/network/simulations/discovery/discovery_test.go b/swarm/network/simulations/discovery/discovery_test.go index 61acd84faa..7dd59beacc 100644 --- a/swarm/network/simulations/discovery/discovery_test.go +++ b/swarm/network/simulations/discovery/discovery_test.go @@ -137,7 +137,7 @@ func benchmarkDiscovery(b *testing.B, nodes, conns int) { for i := 0; i < b.N; i++ { result, err := discoverySimulation(nodes, conns, adapters.NewSimAdapter(services)) if err != nil { - b.Fatalf("setting up simulation failed", result) + b.Fatalf("setting up simulation failed: %v", err) } if result.Error != nil { b.Logf("simulation failed: %s", result.Error) diff --git a/swarm/pss/client/client_test.go b/swarm/pss/client/client_test.go index e773018a16..a6a909b136 100644 --- a/swarm/pss/client/client_test.go +++ b/swarm/pss/client/client_test.go @@ -104,7 +104,8 @@ func TestClientHandshake(t *testing.T) { lproto := pss.NewPingProtocol(lpssping) rproto := pss.NewPingProtocol(rpssping) - ctx, _ := context.WithTimeout(context.Background(), time.Second*10) + ctx, cancel := context.WithTimeout(context.Background(), time.Second*10) + defer cancel() err = lpsc.RunProtocol(ctx, lproto) if err != nil { t.Fatal(err) @@ -231,13 +232,14 @@ func newServices() adapters.Services { "pss": func(ctx *adapters.ServiceContext) (node.Service, error) { cachedir, err := ioutil.TempDir("", "pss-cache") if err != nil { - return nil, fmt.Errorf("create pss cache tmpdir failed", "error", err) + return nil, fmt.Errorf("create pss cache tmpdir failed: %v", err) } dpa, err := storage.NewLocalDPA(cachedir) if err != nil { - return nil, fmt.Errorf("local dpa creation failed", "error", err) + return nil, fmt.Errorf("local dpa creation failed: %v", err) } - ctxlocal, _ := context.WithTimeout(context.Background(), time.Second) + ctxlocal, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() keys, err := wapi.NewKeyPair(ctxlocal) privkey, err := w.GetPrivateKey(keys) psparams := pss.NewPssParams(privkey) diff --git a/swarm/pss/handshake.go b/swarm/pss/handshake.go index 95bf79ef5a..15f2a32a00 100644 --- a/swarm/pss/handshake.go +++ b/swarm/pss/handshake.go @@ -268,7 +268,7 @@ func (self *HandshakeController) handler(msg []byte, p *p2p.Peer, asymmetric boo if !asymmetric { if self.symKeyIndex[symkeyid] != nil { if self.symKeyIndex[symkeyid].count >= self.symKeyIndex[symkeyid].limit { - return fmt.Errorf("discarding message using expired key", "symkeyid", symkeyid) + return fmt.Errorf("discarding message using expired key: %s", symkeyid) } self.symKeyIndex[symkeyid].count++ log.Trace("increment symkey recv use", "symsymkeyid", symkeyid, "count", self.symKeyIndex[symkeyid].count, "limit", self.symKeyIndex[symkeyid].limit, "receiver", common.ToHex(crypto.FromECDSAPub(self.pss.PublicKey()))) @@ -457,7 +457,8 @@ func (self *HandshakeAPI) Handshake(pubkeyid string, topic Topic, sync bool, flu return keys, err } if sync { - ctx, _ := context.WithTimeout(context.Background(), self.ctrl.symKeyRequestTimeout) + ctx, cancel := context.WithTimeout(context.Background(), self.ctrl.symKeyRequestTimeout) + defer cancel() select { case keys = <-hsc: log.Trace("sync handshake response receive", "key", keys) diff --git a/swarm/pss/protocol.go b/swarm/pss/protocol.go index 6c5c289559..9f7c0a6cfe 100644 --- a/swarm/pss/protocol.go +++ b/swarm/pss/protocol.go @@ -227,7 +227,7 @@ func (self *Protocol) AddPeer(p *p2p.Peer, run func(*p2p.Peer, p2p.MsgReadWriter } go func() { err := run(p, rw) - log.Warn(fmt.Sprintf("pss vprotocol quit on addr %v topic %v: %v", topic, err)) + log.Warn(fmt.Sprintf("pss vprotocol quit on %v topic %v: %v", p, topic, err)) }() return rw, nil } diff --git a/swarm/pss/protocol_test.go b/swarm/pss/protocol_test.go index 54cd4226d7..b30fc0430d 100644 --- a/swarm/pss/protocol_test.go +++ b/swarm/pss/protocol_test.go @@ -73,11 +73,13 @@ func testProtocol(t *testing.T) { time.Sleep(time.Millisecond * 1000) // replace with hive healthy code lmsgC := make(chan APIMsg) - lctx, _ := context.WithTimeout(context.Background(), time.Second*10) + lctx, cancel := context.WithTimeout(context.Background(), time.Second*10) + defer cancel() lsub, err := clients[0].Subscribe(lctx, "pss", lmsgC, "receive", topic) defer lsub.Unsubscribe() rmsgC := make(chan APIMsg) - rctx, _ := context.WithTimeout(context.Background(), time.Second*10) + rctx, cancel := context.WithTimeout(context.Background(), time.Second*10) + defer cancel() rsub, err := clients[1].Subscribe(rctx, "pss", rmsgC, "receive", topic) defer rsub.Unsubscribe() diff --git a/swarm/pss/pss.go b/swarm/pss/pss.go index 507b4ab655..0cd226ba08 100644 --- a/swarm/pss/pss.go +++ b/swarm/pss/pss.go @@ -501,7 +501,7 @@ func (self *Pss) processSym(envelope *whisper.Envelope) (*whisper.ReceivedMessag func (self *Pss) processAsym(envelope *whisper.Envelope) (*whisper.ReceivedMessage, string, *PssAddress, error) { recvmsg, err := envelope.OpenAsymmetric(self.privateKey) if err != nil { - return nil, "", nil, fmt.Errorf("could not decrypt message: %v", "err", err) + return nil, "", nil, fmt.Errorf("could not decrypt message: %v", err) } // check signature (if signed), strip padding if !recvmsg.Validate() { diff --git a/swarm/pss/pss_test.go b/swarm/pss/pss_test.go index ae03e3cca5..1b26bfb3fe 100644 --- a/swarm/pss/pss_test.go +++ b/swarm/pss/pss_test.go @@ -137,7 +137,8 @@ func TestTopic(t *testing.T) { func TestCache(t *testing.T) { var err error to, _ := hex.DecodeString("08090a0b0c0d0e0f1011121314150001020304050607161718191a1b1c1d1e1f") - ctx, _ := context.WithTimeout(context.Background(), time.Second) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() keys, err := wapi.NewKeyPair(ctx) privkey, err := w.GetPrivateKey(keys) if err != nil { @@ -211,7 +212,8 @@ func TestAddressMatch(t *testing.T) { remoteaddr := []byte("feedbeef") kadparams := network.NewKadParams() kad := network.NewKademlia(localaddr, kadparams) - ctx, _ := context.WithTimeout(context.Background(), time.Second) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() keys, err := wapi.NewKeyPair(ctx) if err != nil { t.Fatalf("Could not generate private key: %v", err) @@ -255,12 +257,14 @@ func TestAddressMatch(t *testing.T) { // set and generate pubkeys and symkeys func TestKeys(t *testing.T) { // make our key and init pss with it - ctx, _ := context.WithTimeout(context.Background(), time.Second) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() ourkeys, err := wapi.NewKeyPair(ctx) if err != nil { t.Fatalf("create 'our' key fail") } - ctx, _ = context.WithTimeout(context.Background(), time.Second) + ctx, cancel = context.WithTimeout(context.Background(), time.Second) + defer cancel() theirkeys, err := wapi.NewKeyPair(ctx) if err != nil { t.Fatalf("create 'their' key fail") @@ -449,12 +453,14 @@ func testSymSend(t *testing.T) { // at this point we've verified that symkeys are saved and match on each peer // now try sending symmetrically encrypted message, both directions lmsgC := make(chan APIMsg) - lctx, _ := context.WithTimeout(context.Background(), time.Second*10) + lctx, cancel := context.WithTimeout(context.Background(), time.Second*10) + defer cancel() lsub, err := clients[0].Subscribe(lctx, "pss", lmsgC, "receive", topic) log.Trace("lsub", "id", lsub) defer lsub.Unsubscribe() rmsgC := make(chan APIMsg) - rctx, _ := context.WithTimeout(context.Background(), time.Second*10) + rctx, cancel := context.WithTimeout(context.Background(), time.Second*10) + defer cancel() rsub, err := clients[1].Subscribe(rctx, "pss", rmsgC, "receive", topic) log.Trace("rsub", "id", rsub) defer rsub.Unsubscribe() @@ -562,12 +568,14 @@ func testAsymSend(t *testing.T) { time.Sleep(time.Millisecond * 500) // replace with hive healthy code lmsgC := make(chan APIMsg) - lctx, _ := context.WithTimeout(context.Background(), time.Second*10) + lctx, cancel := context.WithTimeout(context.Background(), time.Second*10) + defer cancel() lsub, err := clients[0].Subscribe(lctx, "pss", lmsgC, "receive", topic) log.Trace("lsub", "id", lsub) defer lsub.Unsubscribe() rmsgC := make(chan APIMsg) - rctx, _ := context.WithTimeout(context.Background(), time.Second*10) + rctx, cancel := context.WithTimeout(context.Background(), time.Second*10) + defer cancel() rsub, err := clients[1].Subscribe(rctx, "pss", rmsgC, "receive", topic) log.Trace("rsub", "id", rsub) defer rsub.Unsubscribe() @@ -627,7 +635,6 @@ func worker(id int, jobs <-chan Job, rpcs map[discover.NodeID]*rpc.Client, pubke // nodes/msgs/addrbytes/adaptertype // if adaptertype is exec uses execadapter, simadapter otherwise func TestNetwork(t *testing.T) { - t.Skip("skip until proper local benchmark values for stress testing can be determined") t.Run("3/2000/4/sock", testNetwork) t.Run("4/2000/4/sock", testNetwork) t.Run("8/2000/4/sock", testNetwork) @@ -835,7 +842,8 @@ func benchmarkSymKeySend(b *testing.B) { if err != nil { b.Fatalf("benchmark called with invalid msgsize param '%s': %v", msgsizestring[1], err) } - ctx, _ := context.WithTimeout(context.Background(), time.Second) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() keys, err := wapi.NewKeyPair(ctx) privkey, err := w.GetPrivateKey(keys) ps := newTestPss(privkey, nil, nil) @@ -878,7 +886,8 @@ func benchmarkAsymKeySend(b *testing.B) { if err != nil { b.Fatalf("benchmark called with invalid msgsize param '%s': %v", msgsizestring[1], err) } - ctx, _ := context.WithTimeout(context.Background(), time.Second) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() keys, err := wapi.NewKeyPair(ctx) privkey, err := w.GetPrivateKey(keys) ps := newTestPss(privkey, nil, nil) @@ -923,7 +932,8 @@ func benchmarkSymkeyBruteforceChangeaddr(b *testing.B) { } pssmsgs := make([]*PssMsg, 0, keycount) var keyid string - ctx, _ := context.WithTimeout(context.Background(), time.Second) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() keys, err := wapi.NewKeyPair(ctx) privkey, err := w.GetPrivateKey(keys) if cachesize > 0 { @@ -1005,7 +1015,8 @@ func benchmarkSymkeyBruteforceSameaddr(b *testing.B) { } } addr := make([]PssAddress, keycount) - ctx, _ := context.WithTimeout(context.Background(), time.Second) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() keys, err := wapi.NewKeyPair(ctx) privkey, err := w.GetPrivateKey(keys) if cachesize > 0 { @@ -1122,17 +1133,18 @@ func newServices() adapters.Services { pssProtocolName: func(ctx *adapters.ServiceContext) (node.Service, error) { cachedir, err := ioutil.TempDir("", "pss-cache") if err != nil { - return nil, fmt.Errorf("create pss cache tmpdir failed", "error", err) + return nil, fmt.Errorf("create pss cache tmpdir failed: %v", err) } dpa, err := storage.NewLocalDPA(cachedir) if err != nil { - return nil, fmt.Errorf("local dpa creation failed", "error", err) + return nil, fmt.Errorf("local dpa creation failed: %v", err) } // execadapter does not exec init() initTest() - ctxlocal, _ := context.WithTimeout(context.Background(), time.Second) + ctxlocal, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() keys, err := wapi.NewKeyPair(ctxlocal) privkey, err := w.GetPrivateKey(keys) pssp := NewPssParams(privkey)