Merge pull request #1376 from tendermint/1368-unsubscribe-does-not-work

[rpc] unsubscribe does not work

Gopkg.lock

@@ -268,6 +268,7 @@
   version = "v0.7.3"
 
 [[projects]]
+  branch = "develop"
   name = "github.com/tendermint/tmlibs"
   packages = [
     "autofile",
@@ -283,8 +284,7 @@
     "pubsub/query",
     "test"
   ]
-  revision = "24da7009c3d8c019b40ba4287495749e3160caca"
-  version = "v0.7.1"
+  revision = "0f2811441f4cf44b414df16ceae3c0931c74662e"
 
 [[projects]]
   branch = "master"
@@ -381,6 +381,6 @@
 [solve-meta]
   analyzer-name = "dep"
   analyzer-version = 1
-  inputs-digest = "4dca5dbd2d280d093d7c8fc423606ab86d6ad1b241b076a7716c2093b5a09231"
+  inputs-digest = "88f227f33e785669706f12e131610a86fd3c00f5dfd6f22c62d46e0688049726"
   solver-name = "gps-cdcl"
   solver-version = 1

Gopkg.toml

@@ -82,9 +82,11 @@
   source = "github.com/tendermint/go-amino"
   version = "~0.7.3"
 
-[[constraint]]
+[[override]]
+# [[constraint]]
   name = "github.com/tendermint/tmlibs"
-  version = "~0.7.1"
+  # version = "~0.7.1"
+  branch = "develop"
 
 [[constraint]]
   name = "google.golang.org/grpc"

p2p/conn/connection.go

@@ -7,7 +7,6 @@ import (
 	"io"
 	"math"
 	"net"
-	"runtime/debug"
 	"sync/atomic"
 	"time"
 
@@ -230,8 +229,7 @@ func (c *MConnection) flush() {
 // Catch panics, usually caused by remote disconnects.
 func (c *MConnection) _recover() {
 	if r := recover(); r != nil {
-		stack := debug.Stack()
-		err := cmn.StackError{r, stack}
+		err := cmn.ErrorWrap(r, "recovered from panic")
 		c.stopForError(err)
 	}
 }

p2p/conn/secret_connection.go

@@ -20,8 +20,8 @@ import (
 	"golang.org/x/crypto/nacl/secretbox"
 	"golang.org/x/crypto/ripemd160"
 
-	"github.com/tendermint/go-crypto"
-	"github.com/tendermint/go-wire"
+	crypto "github.com/tendermint/go-crypto"
+	wire "github.com/tendermint/go-wire"
 	cmn "github.com/tendermint/tmlibs/common"
 )
 
@@ -200,26 +200,36 @@ func genEphKeys() (ephPub, ephPriv *[32]byte) {
 }
 
 func shareEphPubKey(conn io.ReadWriteCloser, locEphPub *[32]byte) (remEphPub *[32]byte, err error) {
-	var err1, err2 error
-
-	cmn.Parallel(
-		func() {
-			_, err1 = conn.Write(locEphPub[:])
+	// Send our pubkey and receive theirs in tandem.
+	var trs, _ = cmn.Parallel(
+		func(_ int) (val interface{}, err error, abort bool) {
+			var _, err1 = conn.Write(locEphPub[:])
+			if err1 != nil {
+				return nil, err1, true // abort
+			} else {
+				return nil, nil, false
+			}
 		},
-		func() {
-			remEphPub = new([32]byte)
-			_, err2 = io.ReadFull(conn, remEphPub[:])
+		func(_ int) (val interface{}, err error, abort bool) {
+			var _remEphPub [32]byte
+			var _, err2 = io.ReadFull(conn, _remEphPub[:])
+			if err2 != nil {
+				return nil, err2, true // abort
+			} else {
+				return _remEphPub, nil, false
+			}
 		},
 	)
 
-	if err1 != nil {
-		return nil, err1
-	}
-	if err2 != nil {
-		return nil, err2
+	// If error:
+	if trs.FirstError() != nil {
+		err = trs.FirstError()
+		return
 	}
 
-	return remEphPub, nil
+	// Otherwise:
+	var _remEphPub = trs.FirstValue().([32]byte)
+	return &_remEphPub, nil
 }
 
 func computeSharedSecret(remPubKey, locPrivKey *[32]byte) (shrSecret *[32]byte) {
@@ -268,33 +278,42 @@ type authSigMessage struct {
 	Sig crypto.Signature
 }
 
-func shareAuthSignature(sc *SecretConnection, pubKey crypto.PubKey, signature crypto.Signature) (*authSigMessage, error) {
-	var recvMsg authSigMessage
-	var err1, err2 error
-
-	cmn.Parallel(
-		func() {
+func shareAuthSignature(sc *SecretConnection, pubKey crypto.PubKey, signature crypto.Signature) (recvMsg *authSigMessage, err error) {
+	// Send our info and receive theirs in tandem.
+	var trs, _ = cmn.Parallel(
+		func(_ int) (val interface{}, err error, abort bool) {
 			msgBytes := wire.BinaryBytes(authSigMessage{pubKey.Wrap(), signature.Wrap()})
-			_, err1 = sc.Write(msgBytes)
+			var _, err1 = sc.Write(msgBytes)
+			if err1 != nil {
+				return nil, err1, true // abort
+			} else {
+				return nil, nil, false
+			}
 		},
-		func() {
+		func(_ int) (val interface{}, err error, abort bool) {
 			readBuffer := make([]byte, authSigMsgSize)
-			_, err2 = io.ReadFull(sc, readBuffer)
+			var _, err2 = io.ReadFull(sc, readBuffer)
 			if err2 != nil {
-				return
+				return nil, err2, true // abort
 			}
 			n := int(0) // not used.
-			recvMsg = wire.ReadBinary(authSigMessage{}, bytes.NewBuffer(readBuffer), authSigMsgSize, &n, &err2).(authSigMessage)
-		})
+			var _recvMsg = wire.ReadBinary(authSigMessage{}, bytes.NewBuffer(readBuffer), authSigMsgSize, &n, &err2).(authSigMessage)
+			if err2 != nil {
+				return nil, err2, true // abort
+			} else {
+				return _recvMsg, nil, false
+			}
+		},
+	)
 
-	if err1 != nil {
-		return nil, err1
-	}
-	if err2 != nil {
-		return nil, err2
+	// If error:
+	if trs.FirstError() != nil {
+		err = trs.FirstError()
+		return
 	}
 
-	return &recvMsg, nil
+	var _recvMsg = trs.FirstValue().(authSigMessage)
+	return &_recvMsg, nil
 }
 
 //--------------------------------------------------------------------------------

p2p/conn/secret_connection_test.go

@@ -1,9 +1,12 @@
 package conn
 
 import (
+	"fmt"
 	"io"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	crypto "github.com/tendermint/go-crypto"
 	cmn "github.com/tendermint/tmlibs/common"
 )
@@ -36,33 +39,41 @@ func makeSecretConnPair(tb testing.TB) (fooSecConn, barSecConn *SecretConnection
 	barPrvKey := crypto.GenPrivKeyEd25519().Wrap()
 	barPubKey := barPrvKey.PubKey()
 
-	cmn.Parallel(
-		func() {
-			var err error
+	var trs, ok = cmn.Parallel(
+		func(_ int) (val interface{}, err error, abort bool) {
 			fooSecConn, err = MakeSecretConnection(fooConn, fooPrvKey)
 			if err != nil {
 				tb.Errorf("Failed to establish SecretConnection for foo: %v", err)
-				return
+				return nil, err, true
 			}
 			remotePubBytes := fooSecConn.RemotePubKey()
 			if !remotePubBytes.Equals(barPubKey) {
-				tb.Errorf("Unexpected fooSecConn.RemotePubKey.  Expected %v, got %v",
+				err = fmt.Errorf("Unexpected fooSecConn.RemotePubKey.  Expected %v, got %v",
 					barPubKey, fooSecConn.RemotePubKey())
+				tb.Error(err)
+				return nil, err, false
 			}
+			return nil, nil, false
 		},
-		func() {
-			var err error
+		func(_ int) (val interface{}, err error, abort bool) {
 			barSecConn, err = MakeSecretConnection(barConn, barPrvKey)
 			if barSecConn == nil {
 				tb.Errorf("Failed to establish SecretConnection for bar: %v", err)
-				return
+				return nil, err, true
 			}
 			remotePubBytes := barSecConn.RemotePubKey()
 			if !remotePubBytes.Equals(fooPubKey) {
-				tb.Errorf("Unexpected barSecConn.RemotePubKey.  Expected %v, got %v",
+				err = fmt.Errorf("Unexpected barSecConn.RemotePubKey.  Expected %v, got %v",
 					fooPubKey, barSecConn.RemotePubKey())
+				tb.Error(err)
+				return nil, nil, false
 			}
-		})
+			return nil, nil, false
+		},
+	)
+
+	require.Nil(tb, trs.FirstError())
+	require.True(tb, ok, "Unexpected task abortion")
 
 	return
 }
@@ -89,59 +100,76 @@ func TestSecretConnectionReadWrite(t *testing.T) {
 	}
 
 	// A helper that will run with (fooConn, fooWrites, fooReads) and vice versa
-	genNodeRunner := func(nodeConn kvstoreConn, nodeWrites []string, nodeReads *[]string) func() {
-		return func() {
+	genNodeRunner := func(nodeConn kvstoreConn, nodeWrites []string, nodeReads *[]string) cmn.Task {
+		return func(_ int) (interface{}, error, bool) {
 			// Node handskae
 			nodePrvKey := crypto.GenPrivKeyEd25519().Wrap()
 			nodeSecretConn, err := MakeSecretConnection(nodeConn, nodePrvKey)
 			if err != nil {
 				t.Errorf("Failed to establish SecretConnection for node: %v", err)
-				return
+				return nil, err, true
 			}
-			// In parallel, handle reads and writes
-			cmn.Parallel(
-				func() {
+			// In parallel, handle some reads and writes.
+			var trs, ok = cmn.Parallel(
+				func(_ int) (interface{}, error, bool) {
 					// Node writes
 					for _, nodeWrite := range nodeWrites {
 						n, err := nodeSecretConn.Write([]byte(nodeWrite))
 						if err != nil {
 							t.Errorf("Failed to write to nodeSecretConn: %v", err)
-							return
+							return nil, err, true
 						}
 						if n != len(nodeWrite) {
-							t.Errorf("Failed to write all bytes. Expected %v, wrote %v", len(nodeWrite), n)
-							return
+							err = fmt.Errorf("Failed to write all bytes. Expected %v, wrote %v", len(nodeWrite), n)
+							t.Error(err)
+							return nil, err, true
 						}
 					}
 					if err := nodeConn.PipeWriter.Close(); err != nil {
 						t.Error(err)
+						return nil, err, true
 					}
+					return nil, nil, false
 				},
-				func() {
+				func(_ int) (interface{}, error, bool) {
 					// Node reads
 					readBuffer := make([]byte, dataMaxSize)
 					for {
 						n, err := nodeSecretConn.Read(readBuffer)
 						if err == io.EOF {
-							return
+							return nil, nil, false
 						} else if err != nil {
 							t.Errorf("Failed to read from nodeSecretConn: %v", err)
-							return
+							return nil, err, true
 						}
 						*nodeReads = append(*nodeReads, string(readBuffer[:n]))
 					}
 					if err := nodeConn.PipeReader.Close(); err != nil {
 						t.Error(err)
+						return nil, err, true
 					}
-				})
+					return nil, nil, false
+				},
+			)
+			assert.True(t, ok, "Unexpected task abortion")
+
+			// If error:
+			if trs.FirstError() != nil {
+				return nil, trs.FirstError(), true
+			}
+
+			// Otherwise:
+			return nil, nil, false
 		}
 	}
 
 	// Run foo & bar in parallel
-	cmn.Parallel(
+	var trs, ok = cmn.Parallel(
 		genNodeRunner(fooConn, fooWrites, &fooReads),
 		genNodeRunner(barConn, barWrites, &barReads),
 	)
+	require.Nil(t, trs.FirstError())
+	require.True(t, ok, "unexpected task abortion")
 
 	// A helper to ensure that the writes and reads match.
 	// Additionally, small writes (<= dataMaxSize) must be atomically read.

p2p/peer.go

@@ -293,22 +293,20 @@ func (pc *peerConn) HandshakeTimeout(ourNodeInfo NodeInfo, timeout time.Duration
 		return peerNodeInfo, errors.Wrap(err, "Error setting deadline")
 	}
 
-	var err1 error
-	var err2 error
-	cmn.Parallel(
-		func() {
+	var trs, _ = cmn.Parallel(
+		func(_ int) (val interface{}, err error, abort bool) {
 			var n int
-			wire.WriteBinary(&ourNodeInfo, pc.conn, &n, &err1)
+			wire.WriteBinary(&ourNodeInfo, pc.conn, &n, &err)
+			return
 		},
-		func() {
+		func(_ int) (val interface{}, err error, abort bool) {
 			var n int
-			wire.ReadBinary(&peerNodeInfo, pc.conn, MaxNodeInfoSize(), &n, &err2)
-		})
-	if err1 != nil {
-		return peerNodeInfo, errors.Wrap(err1, "Error during handshake/write")
-	}
-	if err2 != nil {
-		return peerNodeInfo, errors.Wrap(err2, "Error during handshake/read")
+			wire.ReadBinary(&peerNodeInfo, pc.conn, MaxNodeInfoSize(), &n, &err)
+			return
+		},
+	)
+	if err := trs.FirstError(); err != nil {
+		return peerNodeInfo, errors.Wrap(err, "Error during handshake")
 	}
 
 	// Remove deadline