Upgrade Go Callisto

.github/CODEOWNERS

@@ -5,5 +5,7 @@ accounts/usbwallet @karalabe
 consensus          @karalabe
 core/              @karalabe @holiman
 eth/               @karalabe
+les/               @zsfelfoldi
+light/             @zsfelfoldi
 mobile/            @karalabe
 p2p/               @fjl @zsfelfoldi

.travis.yml

@@ -3,17 +3,6 @@ go_import_path: github.com/EthereumCommonwealth/go-callisto
 sudo: false
 matrix:
   include:
-    - os: linux
-      dist: trusty
-      sudo: required
-      go: 1.7.x
-      script:
-        - sudo modprobe fuse
-        - sudo chmod 666 /dev/fuse
-        - sudo chown root:$USER /etc/fuse.conf
-        - go run build/ci.go install
-        - go run build/ci.go test -coverage
-
     - os: linux
       dist: trusty
       sudo: required

build/ci.go

@@ -182,13 +182,13 @@ func doInstall(cmdline []string) {
 	// Check Go version. People regularly open issues about compilation
 	// failure with outdated Go. This should save them the trouble.
 	if !strings.Contains(runtime.Version(), "devel") {
-		// Figure out the minor version number since we can't textually compare (1.10 < 1.7)
+		// Figure out the minor version number since we can't textually compare (1.10 < 1.8)
 		var minor int
 		fmt.Sscanf(strings.TrimPrefix(runtime.Version(), "go1."), "%d", &minor)
 
-		if minor < 7 {
+		if minor < 8 {
 			log.Println("You have Go version", runtime.Version())
-			log.Println("go-ethereum requires at least Go version 1.7 and cannot")
+			log.Println("go-ethereum requires at least Go version 1.8 and cannot")
 			log.Println("be compiled with an earlier version. Please upgrade your Go installation.")
 			os.Exit(1)
 		}

cmd/faucet/faucet.go

@@ -533,9 +533,11 @@ func (f *faucet) loop() {
 	}
 	defer sub.Unsubscribe()
 
-	for {
-		select {
-		case head := <-heads:
+	// Start a goroutine to update the state from head notifications in the background
+	update := make(chan *types.Header)
+
+	go func() {
+		for head := range update {
 			// New chain head arrived, query the current stats and stream to clients
 			var (
 				balance *big.Int
@@ -588,6 +590,17 @@ func (f *faucet) loop() {
 				}
 			}
 			f.lock.RUnlock()
+		}
+	}()
+	// Wait for various events and assing to the appropriate background threads
+	for {
+		select {
+		case head := <-heads:
+			// New head arrived, send if for state update if there's none running
+			select {
+			case update <- head:
+			default:
+			}
 
 		case <-f.update:
 			// Pending requests updated, stream to clients

cmd/swarm/manifest.go

@@ -35,7 +35,7 @@ const bzzManifestJSON = "application/bzz-manifest+json"
 func add(ctx *cli.Context) {
 	args := ctx.Args()
 	if len(args) < 3 {
-		utils.Fatalf("Need atleast three arguments <MHASH> <path> <HASH> [<content-type>]")
+		utils.Fatalf("Need at least three arguments <MHASH> <path> <HASH> [<content-type>]")
 	}
 
 	var (
@@ -69,7 +69,7 @@ func update(ctx *cli.Context) {
 
 	args := ctx.Args()
 	if len(args) < 3 {
-		utils.Fatalf("Need atleast three arguments <MHASH> <path> <HASH>")
+		utils.Fatalf("Need at least three arguments <MHASH> <path> <HASH>")
 	}
 
 	var (
@@ -101,7 +101,7 @@ func update(ctx *cli.Context) {
 func remove(ctx *cli.Context) {
 	args := ctx.Args()
 	if len(args) < 2 {
-		utils.Fatalf("Need atleast two arguments <MHASH> <path>")
+		utils.Fatalf("Need at least two arguments <MHASH> <path>")
 	}
 
 	var (

cmd/wnode/main.go

@@ -22,6 +22,7 @@ package main
 import (
 	"bufio"
 	"crypto/ecdsa"
+	crand "crypto/rand"
 	"crypto/sha512"
 	"encoding/binary"
 	"encoding/hex"
@@ -48,6 +49,7 @@ import (
 )
 
 const quitCommand = "~Q"
+const entropySize = 32
 
 // singletons
 var (
@@ -55,6 +57,7 @@ var (
 	shh        *whisper.Whisper
 	done       chan struct{}
 	mailServer mailserver.WMailServer
+	entropy    [entropySize]byte
 
 	input = bufio.NewReader(os.Stdin)
 )
@@ -83,6 +86,7 @@ var (
 	asymmetricMode = flag.Bool("asym", false, "use asymmetric encryption")
 	generateKey    = flag.Bool("generatekey", false, "generate and show the private key")
 	fileExMode     = flag.Bool("fileexchange", false, "file exchange mode")
+	fileReader     = flag.Bool("filereader", false, "load and decrypt messages saved as files, display as plain text")
 	testMode       = flag.Bool("test", false, "use of predefined parameters for diagnostics (password, etc.)")
 	echoMode       = flag.Bool("echo", false, "echo mode: prints some arguments for diagnostics")
 
@@ -274,6 +278,11 @@ func initialize() {
 			TrustedNodes:   peers,
 		},
 	}
+
+	_, err = crand.Read(entropy[:])
+	if err != nil {
+		utils.Fatalf("crypto/rand failed: %s", err)
+	}
 }
 
 func startServer() {
@@ -425,6 +434,8 @@ func run() {
 		requestExpiredMessagesLoop()
 	} else if *fileExMode {
 		sendFilesLoop()
+	} else if *fileReader {
+		fileReaderLoop()
 	} else {
 		sendLoop()
 	}
@@ -475,6 +486,40 @@ func sendFilesLoop() {
 	}
 }
 
+func fileReaderLoop() {
+	watcher1 := shh.GetFilter(symFilterID)
+	watcher2 := shh.GetFilter(asymFilterID)
+	if watcher1 == nil && watcher2 == nil {
+		fmt.Println("Error: neither symmetric nor asymmetric filter is installed")
+		close(done)
+		return
+	}
+
+	for {
+		s := scanLine("")
+		if s == quitCommand {
+			fmt.Println("Quit command received")
+			close(done)
+			return
+		}
+		raw, err := ioutil.ReadFile(s)
+		if err != nil {
+			fmt.Printf(">>> Error: %s \n", err)
+		} else {
+			env := whisper.Envelope{Data: raw} // the topic is zero
+			msg := env.Open(watcher1)          // force-open envelope regardless of the topic
+			if msg == nil {
+				msg = env.Open(watcher2)
+			}
+			if msg == nil {
+				fmt.Printf(">>> Error: failed to decrypt the message \n")
+			} else {
+				printMessageInfo(msg)
+			}
+		}
+	}
+}
+
 func scanLine(prompt string) string {
 	if len(prompt) > 0 {
 		fmt.Print(prompt)
@@ -594,27 +639,30 @@ func writeMessageToFile(dir string, msg *whisper.ReceivedMessage) {
 		address = crypto.PubkeyToAddress(*msg.Src)
 	}
 
-	if whisper.IsPubKeyEqual(msg.Src, &asymKey.PublicKey) {
-		// message from myself: don't save, only report
-		fmt.Printf("\n%s <%x>: message received: '%s'\n", timestamp, address, name)
-	} else if len(dir) > 0 {
+	// this is a sample code; uncomment if you don't want to save your own messages.
+	//if whisper.IsPubKeyEqual(msg.Src, &asymKey.PublicKey) {
+	//	fmt.Printf("\n%s <%x>: message from myself received, not saved: '%s'\n", timestamp, address, name)
+	//	return
+	//}
+
+	if len(dir) > 0 {
 		fullpath := filepath.Join(dir, name)
-		err := ioutil.WriteFile(fullpath, msg.Payload, 0644)
+		err := ioutil.WriteFile(fullpath, msg.Raw, 0644)
 		if err != nil {
 			fmt.Printf("\n%s {%x}: message received but not saved: %s\n", timestamp, address, err)
 		} else {
-			fmt.Printf("\n%s {%x}: message received and saved as '%s' (%d bytes)\n", timestamp, address, name, len(msg.Payload))
+			fmt.Printf("\n%s {%x}: message received and saved as '%s' (%d bytes)\n", timestamp, address, name, len(msg.Raw))
 		}
 	} else {
-		fmt.Printf("\n%s {%x}: big message received (%d bytes), but not saved: %s\n", timestamp, address, len(msg.Payload), name)
+		fmt.Printf("\n%s {%x}: message received (%d bytes), but not saved: %s\n", timestamp, address, len(msg.Raw), name)
 	}
 }
 
 func requestExpiredMessagesLoop() {
-	var key, peerID []byte
+	var key, peerID, bloom []byte
 	var timeLow, timeUpp uint32
 	var t string
-	var xt, empty whisper.TopicType
+	var xt whisper.TopicType
 
 	keyID, err := shh.AddSymKeyFromPassword(msPassword)
 	if err != nil {
@@ -637,18 +685,19 @@ func requestExpiredMessagesLoop() {
 				utils.Fatalf("Failed to parse the topic: %s", err)
 			}
 			xt = whisper.BytesToTopic(x)
+			bloom = whisper.TopicToBloom(xt)
+			obfuscateBloom(bloom)
+		} else {
+			bloom = whisper.MakeFullNodeBloom()
 		}
 		if timeUpp == 0 {
 			timeUpp = 0xFFFFFFFF
 		}
 
-		data := make([]byte, 8+whisper.TopicLength)
+		data := make([]byte, 8, 8+whisper.BloomFilterSize)
 		binary.BigEndian.PutUint32(data, timeLow)
 		binary.BigEndian.PutUint32(data[4:], timeUpp)
-		copy(data[8:], xt[:])
-		if xt == empty {
-			data = data[:8]
-		}
+		data = append(data, bloom...)
 
 		var params whisper.MessageParams
 		params.PoW = *argServerPoW
@@ -682,3 +731,20 @@ func extractIDFromEnode(s string) []byte {
 	}
 	return n.ID[:]
 }
+
+// obfuscateBloom adds 16 random bits to the the bloom
+// filter, in order to obfuscate the containing topics.
+// it does so deterministically within every session.
+// despite additional bits, it will match on average
+// 32000 times less messages than full node's bloom filter.
+func obfuscateBloom(bloom []byte) {
+	const half = entropySize / 2
+	for i := 0; i < half; i++ {
+		x := int(entropy[i])
+		if entropy[half+i] < 128 {
+			x += 256
+		}
+
+		bloom[x/8] = 1 << uint(x%8) // set the bit number X
+	}
+}

consensus/ethash/algorithm.go

@@ -19,6 +19,7 @@ package ethash
 import (
 	"encoding/binary"
 	"hash"
+	"math/big"
 	"reflect"
 	"runtime"
 	"sync"
@@ -47,6 +48,48 @@ const (
 	loopAccesses       = 64      // Number of accesses in hashimoto loop
 )
 
+// cacheSize returns the size of the ethash verification cache that belongs to a certain
+// block number.
+func cacheSize(block uint64) uint64 {
+	epoch := int(block / epochLength)
+	if epoch < maxEpoch {
+		return cacheSizes[epoch]
+	}
+	return calcCacheSize(epoch)
+}
+
+// calcCacheSize calculates the cache size for epoch. The cache size grows linearly,
+// however, we always take the highest prime below the linearly growing threshold in order
+// to reduce the risk of accidental regularities leading to cyclic behavior.
+func calcCacheSize(epoch int) uint64 {
+	size := cacheInitBytes + cacheGrowthBytes*uint64(epoch) - hashBytes
+	for !new(big.Int).SetUint64(size / hashBytes).ProbablyPrime(1) { // Always accurate for n < 2^64
+		size -= 2 * hashBytes
+	}
+	return size
+}
+
+// datasetSize returns the size of the ethash mining dataset that belongs to a certain
+// block number.
+func datasetSize(block uint64) uint64 {
+	epoch := int(block / epochLength)
+	if epoch < maxEpoch {
+		return datasetSizes[epoch]
+	}
+	return calcDatasetSize(epoch)
+}
+
+// calcDatasetSize calculates the dataset size for epoch. The dataset size grows linearly,
+// however, we always take the highest prime below the linearly growing threshold in order
+// to reduce the risk of accidental regularities leading to cyclic behavior.
+func calcDatasetSize(epoch int) uint64 {
+	size := datasetInitBytes + datasetGrowthBytes*uint64(epoch) - mixBytes
+	for !new(big.Int).SetUint64(size / mixBytes).ProbablyPrime(1) { // Always accurate for n < 2^64
+		size -= 2 * mixBytes
+	}
+	return size
+}
+
 // hasher is a repetitive hasher allowing the same hash data structures to be
 // reused between hash runs instead of requiring new ones to be created.
 type hasher func(dest []byte, data []byte)

consensus/ethash/algorithm_go1.7.go

@@ -1,47 +0,0 @@
-// Copyright 2017 The go-ethereum Authors
-// This file is part of the go-ethereum library.
-//
-// The go-ethereum library is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Lesser General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// The go-ethereum library is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-// GNU Lesser General Public License for more details.
-//
-// You should have received a copy of the GNU Lesser General Public License
-// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
-
-// +build !go1.8
-
-package ethash
-
-// cacheSize calculates and returns the size of the ethash verification cache that
-// belongs to a certain block number. The cache size grows linearly, however, we
-// always take the highest prime below the linearly growing threshold in order to
-// reduce the risk of accidental regularities leading to cyclic behavior.
-func cacheSize(block uint64) uint64 {
-	// If we have a pre-generated value, use that
-	epoch := int(block / epochLength)
-	if epoch < maxEpoch {
-		return cacheSizes[epoch]
-	}
-	// We don't have a way to verify primes fast before Go 1.8
-	panic("fast prime testing unsupported in Go < 1.8")
-}
-
-// datasetSize calculates and returns the size of the ethash mining dataset that
-// belongs to a certain block number. The dataset size grows linearly, however, we
-// always take the highest prime below the linearly growing threshold in order to
-// reduce the risk of accidental regularities leading to cyclic behavior.
-func datasetSize(block uint64) uint64 {
-	// If we have a pre-generated value, use that
-	epoch := int(block / epochLength)
-	if epoch < maxEpoch {
-		return datasetSizes[epoch]
-	}
-	// We don't have a way to verify primes fast before Go 1.8
-	panic("fast prime testing unsupported in Go < 1.8")
-}

consensus/ethash/algorithm_go1.8.go

@@ -1,63 +0,0 @@
-// Copyright 2017 The go-ethereum Authors
-// This file is part of the go-ethereum library.
-//
-// The go-ethereum library is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Lesser General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// The go-ethereum library is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-// GNU Lesser General Public License for more details.
-//
-// You should have received a copy of the GNU Lesser General Public License
-// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
-
-// +build go1.8
-
-package ethash
-
-import "math/big"
-
-// cacheSize returns the size of the ethash verification cache that belongs to a certain
-// block number.
-func cacheSize(block uint64) uint64 {
-	epoch := int(block / epochLength)
-	if epoch < maxEpoch {
-		return cacheSizes[epoch]
-	}
-	return calcCacheSize(epoch)
-}
-
-// calcCacheSize calculates the cache size for epoch. The cache size grows linearly,
-// however, we always take the highest prime below the linearly growing threshold in order
-// to reduce the risk of accidental regularities leading to cyclic behavior.
-func calcCacheSize(epoch int) uint64 {
-	size := cacheInitBytes + cacheGrowthBytes*uint64(epoch) - hashBytes
-	for !new(big.Int).SetUint64(size / hashBytes).ProbablyPrime(1) { // Always accurate for n < 2^64
-		size -= 2 * hashBytes
-	}
-	return size
-}
-
-// datasetSize returns the size of the ethash mining dataset that belongs to a certain
-// block number.
-func datasetSize(block uint64) uint64 {
-	epoch := int(block / epochLength)
-	if epoch < maxEpoch {
-		return datasetSizes[epoch]
-	}
-	return calcDatasetSize(epoch)
-}
-
-// calcDatasetSize calculates the dataset size for epoch. The dataset size grows linearly,
-// however, we always take the highest prime below the linearly growing threshold in order
-// to reduce the risk of accidental regularities leading to cyclic behavior.
-func calcDatasetSize(epoch int) uint64 {
-	size := datasetInitBytes + datasetGrowthBytes*uint64(epoch) - mixBytes
-	for !new(big.Int).SetUint64(size / mixBytes).ProbablyPrime(1) { // Always accurate for n < 2^64
-		size -= 2 * mixBytes
-	}
-	return size
-}

consensus/ethash/algorithm_go1.8_test.go

@@ -1,37 +0,0 @@
-// Copyright 2017 The go-ethereum Authors
-// This file is part of the go-ethereum library.
-//
-// The go-ethereum library is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Lesser General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// The go-ethereum library is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-// GNU Lesser General Public License for more details.
-//
-// You should have received a copy of the GNU Lesser General Public License
-// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
-
-// +build go1.8
-
-package ethash
-
-import "testing"
-
-// Tests whether the dataset size calculator works correctly by cross checking the
-// hard coded lookup table with the value generated by it.
-func TestSizeCalculations(t *testing.T) {
-	// Verify all the cache and dataset sizes from the lookup table.
-	for epoch, want := range cacheSizes {
-		if size := calcCacheSize(epoch); size != want {
-			t.Errorf("cache %d: cache size mismatch: have %d, want %d", epoch, size, want)
-		}
-	}
-	for epoch, want := range datasetSizes {
-		if size := calcDatasetSize(epoch); size != want {
-			t.Errorf("dataset %d: dataset size mismatch: have %d, want %d", epoch, size, want)
-		}
-	}
-}

consensus/ethash/algorithm_test.go

@@ -30,6 +30,22 @@ import (
 	"github.com/EthereumCommonwealth/go-callisto/core/types"
 )
 
+// Tests whether the dataset size calculator works correctly by cross checking the
+// hard coded lookup table with the value generated by it.
+func TestSizeCalculations(t *testing.T) {
+	// Verify all the cache and dataset sizes from the lookup table.
+	for epoch, want := range cacheSizes {
+		if size := calcCacheSize(epoch); size != want {
+			t.Errorf("cache %d: cache size mismatch: have %d, want %d", epoch, size, want)
+		}
+	}
+	for epoch, want := range datasetSizes {
+		if size := calcDatasetSize(epoch); size != want {
+			t.Errorf("dataset %d: dataset size mismatch: have %d, want %d", epoch, size, want)
+		}
+	}
+}
+
 // Tests that verification caches can be correctly generated.
 func TestCacheGeneration(t *testing.T) {
 	tests := []struct {

consensus/ethash/consensus.go

@@ -55,7 +55,6 @@ var (
 	errDuplicateUncle    = errors.New("duplicate uncle")
 	errUncleIsAncestor   = errors.New("uncle is ancestor")
 	errDanglingUncle     = errors.New("uncle's parent is not ancestor")
-	errNonceOutOfRange   = errors.New("nonce out of range")
 	errInvalidDifficulty = errors.New("non-positive difficulty")
 	errInvalidMixDigest  = errors.New("invalid mix digest")
 	errInvalidPoW        = errors.New("invalid proof-of-work")
@@ -527,18 +526,13 @@ func (ethash *Ethash) VerifySeal(chain consensus.ChainReader, header *types.Head
 	if ethash.shared != nil {
 		return ethash.shared.VerifySeal(chain, header)
 	}
-	// Sanity check that the block number is below the lookup table size (60M blocks)
-	number := header.Number.Uint64()
-	if number/epochLength >= maxEpoch {
-		// Go < 1.7 cannot calculate new cache/dataset sizes (no fast prime check)
-		return errNonceOutOfRange
-	}
 	// Ensure that we have a valid difficulty for the block
 	if header.Difficulty.Sign() <= 0 {
 		return errInvalidDifficulty
 	}
-
 	// Recompute the digest and PoW value and verify against the header
+	number := header.Number.Uint64()
+
 	cache := ethash.cache(number)
 	size := datasetSize(number)
 	if ethash.config.PowMode == ModeTest {

core/asm/lexer.go

@@ -206,7 +206,7 @@ func lexLine(l *lexer) stateFn {
 			return lexComment
 		case isSpace(r):
 			l.ignore()
-		case isAlphaNumeric(r) || r == '_':
+		case isLetter(r) || r == '_':
 			return lexElement
 		case isNumber(r):
 			return lexNumber
@@ -278,7 +278,7 @@ func lexElement(l *lexer) stateFn {
 	return lexLine
 }
 
-func isAlphaNumeric(t rune) bool {
+func isLetter(t rune) bool {
 	return unicode.IsLetter(t)
 }
 

core/database_util.go

@@ -47,6 +47,7 @@ var (
 	headHeaderKey = []byte("LastHeader")
 	headBlockKey  = []byte("LastBlock")
 	headFastKey   = []byte("LastFast")
+	trieSyncKey   = []byte("TrieSync")
 
 	// Data item prefixes (use single byte to avoid mixing data types, avoid `i`).
 	headerPrefix        = []byte("h") // headerPrefix + num (uint64 big endian) + hash -> header
@@ -146,6 +147,16 @@ func GetHeadFastBlockHash(db DatabaseReader) common.Hash {
 	return common.BytesToHash(data)
 }
 
+// GetTrieSyncProgress retrieves the number of tries nodes fast synced to allow
+// reportinc correct numbers across restarts.
+func GetTrieSyncProgress(db DatabaseReader) uint64 {
+	data, _ := db.Get(trieSyncKey)
+	if len(data) == 0 {
+		return 0
+	}
+	return new(big.Int).SetBytes(data).Uint64()
+}
+
 // GetHeaderRLP retrieves a block header in its raw RLP database encoding, or nil
 // if the header's not found.
 func GetHeaderRLP(db DatabaseReader, hash common.Hash, number uint64) rlp.RawValue {
@@ -374,6 +385,15 @@ func WriteHeadFastBlockHash(db ethdb.Putter, hash common.Hash) error {
 	return nil
 }
 
+// WriteTrieSyncProgress stores the fast sync trie process counter to support
+// retrieving it across restarts.
+func WriteTrieSyncProgress(db ethdb.Putter, count uint64) error {
+	if err := db.Put(trieSyncKey, new(big.Int).SetUint64(count).Bytes()); err != nil {
+		log.Crit("Failed to store fast sync trie progress", "err", err)
+	}
+	return nil
+}
+
 // WriteHeader serializes a block header into the database.
 func WriteHeader(db ethdb.Putter, header *types.Header) error {
 	data, err := rlp.EncodeToBytes(header)

eth/downloader/downloader.go

@@ -27,6 +27,7 @@ import (
 
 	ethereum "github.com/EthereumCommonwealth/go-callisto"
 	"github.com/EthereumCommonwealth/go-callisto/common"
+	"github.com/EthereumCommonwealth/go-callisto/core"
 	"github.com/EthereumCommonwealth/go-callisto/core/types"
 	"github.com/EthereumCommonwealth/go-callisto/ethdb"
 	"github.com/EthereumCommonwealth/go-callisto/event"
@@ -221,6 +222,9 @@ func New(mode SyncMode, stateDb ethdb.Database, mux *event.TypeMux, chain BlockC
 		quitCh:         make(chan struct{}),
 		stateCh:        make(chan dataPack),
 		stateSyncStart: make(chan *stateSync),
+		syncStatsState: stateSyncStats{
+			processed: core.GetTrieSyncProgress(stateDb),
+		},
 		trackStateReq: make(chan *stateReq),
 	}
 	go dl.qosTuner()

eth/downloader/statesync.go

@@ -23,6 +23,7 @@ import (
 	"time"
 
 	"github.com/EthereumCommonwealth/go-callisto/common"
+	"github.com/EthereumCommonwealth/go-callisto/core"
 	"github.com/EthereumCommonwealth/go-callisto/core/state"
 	"github.com/EthereumCommonwealth/go-callisto/crypto/sha3"
 	"github.com/EthereumCommonwealth/go-callisto/ethdb"
@@ -466,4 +467,7 @@ func (s *stateSync) updateStats(written, duplicate, unexpected int, duration tim
 	if written > 0 || duplicate > 0 || unexpected > 0 {
 		log.Info("Imported new state entries", "count", written, "elapsed", common.PrettyDuration(duration), "processed", s.d.syncStatsState.processed, "pending", s.d.syncStatsState.pending, "retry", len(s.tasks), "duplicate", s.d.syncStatsState.duplicate, "unexpected", s.d.syncStatsState.unexpected)
 	}
+	if written > 0 {
+		core.WriteTrieSyncProgress(s.d.stateDB, s.d.syncStatsState.processed)
+	}
 }

internal/debug/api.go

@@ -140,10 +140,9 @@ func (h *HandlerT) GoTrace(file string, nsec uint) error {
 	return nil
 }
 
-// BlockProfile turns on CPU profiling for nsec seconds and writes
-// profile data to file. It uses a profile rate of 1 for most accurate
-// information. If a different rate is desired, set the rate
-// and write the profile manually.
+// BlockProfile turns on goroutine profiling for nsec seconds and writes profile data to
+// file. It uses a profile rate of 1 for most accurate information. If a different rate is
+// desired, set the rate and write the profile manually.
 func (*HandlerT) BlockProfile(file string, nsec uint) error {
 	runtime.SetBlockProfileRate(1)
 	time.Sleep(time.Duration(nsec) * time.Second)
@@ -162,6 +161,26 @@ func (*HandlerT) WriteBlockProfile(file string) error {
 	return writeProfile("block", file)
 }
 
+// MutexProfile turns on mutex profiling for nsec seconds and writes profile data to file.
+// It uses a profile rate of 1 for most accurate information. If a different rate is
+// desired, set the rate and write the profile manually.
+func (*HandlerT) MutexProfile(file string, nsec uint) error {
+	runtime.SetMutexProfileFraction(1)
+	time.Sleep(time.Duration(nsec) * time.Second)
+	defer runtime.SetMutexProfileFraction(0)
+	return writeProfile("mutex", file)
+}
+
+// SetMutexProfileFraction sets the rate of mutex profiling.
+func (*HandlerT) SetMutexProfileFraction(rate int) {
+	runtime.SetMutexProfileFraction(rate)
+}
+
+// WriteMutexProfile writes a goroutine blocking profile to the given file.
+func (*HandlerT) WriteMutexProfile(file string) error {
+	return writeProfile("mutex", file)
+}
+
 // WriteMemProfile writes an allocation profile to the given file.
 // Note that the profiling rate cannot be set through the API,
 // it must be set on the command line.

internal/ethapi/api.go

@@ -1043,7 +1043,7 @@ func (s *PublicTransactionPoolAPI) GetTransactionReceipt(ctx context.Context, ha
 		return nil, err
 	}
 	if len(receipts) <= int(index) {
-		return nil, errors.New("unknown receipt")
+		return nil, nil
 	}
 	receipt := receipts[index]
 

internal/web3ext/web3ext.go

@@ -307,6 +307,21 @@ web3._extend({
 			call: 'debug_writeBlockProfile',
 			params: 1
 		}),
+		new web3._extend.Method({
+			name: 'mutexProfile',
+			call: 'debug_mutexProfile',
+			params: 2
+		}),
+		new web3._extend.Method({
+			name: 'setMutexProfileRate',
+			call: 'debug_setMutexProfileRate',
+			params: 1
+		}),
+		new web3._extend.Method({
+			name: 'writeMutexProfile',
+			call: 'debug_writeMutexProfile',
+			params: 1
+		}),
 		new web3._extend.Method({
 			name: 'writeMemProfile',
 			call: 'debug_writeMemProfile',

light/postprocess.go

@@ -58,18 +58,18 @@ type trustedCheckpoint struct {
 var (
 	mainnetCheckpoint = trustedCheckpoint{
 		name:          "mainnet",
-		sectionIdx:    153,
-		sectionHead:   common.HexToHash("04c2114a8cbe49ba5c37a03cc4b4b8d3adfc0bd2c78e0e726405dd84afca1d63"),
-		chtRoot:       common.HexToHash("d7ec603e5d30b567a6e894ee7704e4603232f206d3e5a589794cec0c57bf318e"),
-		bloomTrieRoot: common.HexToHash("0b139b8fb692e21f663ff200da287192201c28ef5813c1ac6ba02a0a4799eef9"),
+		sectionIdx:    157,
+		sectionHead:   common.HexToHash("1963c080887ca7f406c2bb114293eea83e54f783f94df24b447f7e3b6317c747"),
+		chtRoot:       common.HexToHash("42abc436567dfb678a38fa6a9f881aa4c8a4cc8eaa2def08359292c3d0bd48ec"),
+		bloomTrieRoot: common.HexToHash("281c9f8fb3cb8b37ae45e9907ef8f3b19cd22c54e297c2d6c09c1db1593dce42"),
 	}
 
 	ropstenCheckpoint = trustedCheckpoint{
 		name:          "ropsten",
-		sectionIdx:    79,
-		sectionHead:   common.HexToHash("1b1ba890510e06411fdee9bb64ca7705c56a1a4ce3559ddb34b3680c526cb419"),
-		chtRoot:       common.HexToHash("71d60207af74e5a22a3e1cfbfc89f9944f91b49aa980c86fba94d568369eaf44"),
-		bloomTrieRoot: common.HexToHash("70aca4b3b6d08dde8704c95cedb1420394453c1aec390947751e69ff8c436360"),
+		sectionIdx:    83,
+		sectionHead:   common.HexToHash("3ca623586bc0da35f1fc8d9b6b55950f3b1f69be9c6501846a2df672adb61236"),
+		chtRoot:       common.HexToHash("8f08ec7783969768c6ef06e5fe3398223cbf4ae2907b676da7b6fe6c7f55b059"),
+		bloomTrieRoot: common.HexToHash("02d86d3c6a87f8f8a92c2a59bbba2132ff6f9f61b0915a5dc28a9d8279219fd0"),
 	}
 )
 

metrics/metrics_test.go

@@ -6,6 +6,7 @@ import (
 	"log"
 	"sync"
 	"testing"
+	"time"
 )
 
 const FANOUT = 128
@@ -114,7 +115,7 @@ func Example() {
 
 	// Threadsafe registration
 	t := GetOrRegisterTimer("db.get.latency", nil)
-	t.Time(func() {})
+	t.Time(func() { time.Sleep(10 * time.Millisecond) })
 	t.Update(1)
 
 	fmt.Println(c.Count())

metrics/timer_test.go

@@ -47,8 +47,8 @@ func TestTimerStop(t *testing.T) {
 func TestTimerFunc(t *testing.T) {
 	tm := NewTimer()
 	tm.Time(func() { time.Sleep(50e6) })
-	if max := tm.Max(); 45e6 > max || max > 55e6 {
-		t.Errorf("tm.Max(): 45e6 > %v || %v > 55e6\n", max, max)
+	if max := tm.Max(); 35e6 > max || max > 95e6 {
+		t.Errorf("tm.Max(): 35e6 > %v || %v > 95e6\n", max, max)
 	}
 }
 

whisper/mailserver/mailserver.go

@@ -17,7 +17,6 @@
 package mailserver
 
 import (
-	"bytes"
 	"encoding/binary"
 	"fmt"
 
@@ -108,17 +107,16 @@ func (s *WMailServer) DeliverMail(peer *whisper.Peer, request *whisper.Envelope)
 		return
 	}
 
-	ok, lower, upper, topic := s.validateRequest(peer.ID(), request)
+	ok, lower, upper, bloom := s.validateRequest(peer.ID(), request)
 	if ok {
-		s.processRequest(peer, lower, upper, topic)
+		s.processRequest(peer, lower, upper, bloom)
 	}
 }
 
-func (s *WMailServer) processRequest(peer *whisper.Peer, lower, upper uint32, topic whisper.TopicType) []*whisper.Envelope {
+func (s *WMailServer) processRequest(peer *whisper.Peer, lower, upper uint32, bloom []byte) []*whisper.Envelope {
 	ret := make([]*whisper.Envelope, 0)
 	var err error
 	var zero common.Hash
-	var empty whisper.TopicType
 	kl := NewDbKey(lower, zero)
 	ku := NewDbKey(upper, zero)
 	i := s.db.NewIterator(&util.Range{Start: kl.raw, Limit: ku.raw}, nil)
@@ -131,7 +129,7 @@ func (s *WMailServer) processRequest(peer *whisper.Peer, lower, upper uint32, to
 			log.Error(fmt.Sprintf("RLP decoding failed: %s", err))
 		}
 
-		if topic == empty || envelope.Topic == topic {
+		if whisper.BloomFilterMatch(bloom, envelope.Bloom()) {
 			if peer == nil {
 				// used for test purposes
 				ret = append(ret, &envelope)
@@ -153,39 +151,45 @@ func (s *WMailServer) processRequest(peer *whisper.Peer, lower, upper uint32, to
 	return ret
 }
 
-func (s *WMailServer) validateRequest(peerID []byte, request *whisper.Envelope) (bool, uint32, uint32, whisper.TopicType) {
-	var topic whisper.TopicType
+func (s *WMailServer) validateRequest(peerID []byte, request *whisper.Envelope) (bool, uint32, uint32, []byte) {
 	if s.pow > 0.0 && request.PoW() < s.pow {
-		return false, 0, 0, topic
+		return false, 0, 0, nil
 	}
 
 	f := whisper.Filter{KeySym: s.key}
 	decrypted := request.Open(&f)
 	if decrypted == nil {
 		log.Warn(fmt.Sprintf("Failed to decrypt p2p request"))
-		return false, 0, 0, topic
-	}
-
-	if len(decrypted.Payload) < 8 {
-		log.Warn(fmt.Sprintf("Undersized p2p request"))
-		return false, 0, 0, topic
+		return false, 0, 0, nil
 	}
 
 	src := crypto.FromECDSAPub(decrypted.Src)
 	if len(src)-len(peerID) == 1 {
 		src = src[1:]
 	}
-	if !bytes.Equal(peerID, src) {
+
+	// if you want to check the signature, you can do it here. e.g.:
+	// if !bytes.Equal(peerID, src) {
+	if src == nil {
 		log.Warn(fmt.Sprintf("Wrong signature of p2p request"))
-		return false, 0, 0, topic
+		return false, 0, 0, nil
+	}
+
+	var bloom []byte
+	payloadSize := len(decrypted.Payload)
+	if payloadSize < 8 {
+		log.Warn(fmt.Sprintf("Undersized p2p request"))
+		return false, 0, 0, nil
+	} else if payloadSize == 8 {
+		bloom = whisper.MakeFullNodeBloom()
+	} else if payloadSize < 8+whisper.BloomFilterSize {
+		log.Warn(fmt.Sprintf("Undersized bloom filter in p2p request"))
+		return false, 0, 0, nil
+	} else {
+		bloom = decrypted.Payload[8 : 8+whisper.BloomFilterSize]
 	}
 
 	lower := binary.BigEndian.Uint32(decrypted.Payload[:4])
 	upper := binary.BigEndian.Uint32(decrypted.Payload[4:8])
-
-	if len(decrypted.Payload) >= 8+whisper.TopicLength {
-		topic = whisper.BytesToTopic(decrypted.Payload[8:])
-	}
-
-	return true, lower, upper, topic
+	return true, lower, upper, bloom
 }

whisper/mailserver/server_test.go

@@ -17,6 +17,7 @@
 package mailserver
 
 import (
+	"bytes"
 	"crypto/ecdsa"
 	"encoding/binary"
 	"io/ioutil"
@@ -61,7 +62,7 @@ func generateEnvelope(t *testing.T) *whisper.Envelope {
 	h := crypto.Keccak256Hash([]byte("test sample data"))
 	params := &whisper.MessageParams{
 		KeySym:   h[:],
-		Topic:    whisper.TopicType{},
+		Topic:    whisper.TopicType{0x1F, 0x7E, 0xA1, 0x7F},
 		Payload:  []byte("test payload"),
 		PoW:      powRequirement,
 		WorkTime: 2,
@@ -121,6 +122,7 @@ func deliverTest(t *testing.T, server *WMailServer, env *whisper.Envelope) {
 		upp:   birth + 1,
 		key:   testPeerID,
 	}
+
 	singleRequest(t, server, env, p, true)
 
 	p.low, p.upp = birth+1, 0xffffffff
@@ -131,14 +133,14 @@ func deliverTest(t *testing.T, server *WMailServer, env *whisper.Envelope) {
 
 	p.low = birth - 1
 	p.upp = birth + 1
-	p.topic[0]++
+	p.topic[0] = 0xFF
 	singleRequest(t, server, env, p, false)
 }
 
 func singleRequest(t *testing.T, server *WMailServer, env *whisper.Envelope, p *ServerTestParams, expect bool) {
 	request := createRequest(t, p)
 	src := crypto.FromECDSAPub(&p.key.PublicKey)
-	ok, lower, upper, topic := server.validateRequest(src, request)
+	ok, lower, upper, bloom := server.validateRequest(src, request)
 	if !ok {
 		t.Fatalf("request validation failed, seed: %d.", seed)
 	}
@@ -148,12 +150,13 @@ func singleRequest(t *testing.T, server *WMailServer, env *whisper.Envelope, p *
 	if upper != p.upp {
 		t.Fatalf("request validation failed (upper bound), seed: %d.", seed)
 	}
-	if topic != p.topic {
+	expectedBloom := whisper.TopicToBloom(p.topic)
+	if !bytes.Equal(bloom, expectedBloom) {
 		t.Fatalf("request validation failed (topic), seed: %d.", seed)
 	}
 
 	var exist bool
-	mail := server.processRequest(nil, p.low, p.upp, p.topic)
+	mail := server.processRequest(nil, p.low, p.upp, bloom)
 	for _, msg := range mail {
 		if msg.Hash() == env.Hash() {
 			exist = true
@@ -166,17 +169,19 @@ func singleRequest(t *testing.T, server *WMailServer, env *whisper.Envelope, p *
 	}
 
 	src[0]++
-	ok, lower, upper, topic = server.validateRequest(src, request)
-	if ok {
-		t.Fatalf("request validation false positive, seed: %d (lower: %d, upper: %d).", seed, lower, upper)
+	ok, lower, upper, bloom = server.validateRequest(src, request)
+	if !ok {
+		// request should be valid regardless of signature
+		t.Fatalf("request validation false negative, seed: %d (lower: %d, upper: %d).", seed, lower, upper)
 	}
 }
 
 func createRequest(t *testing.T, p *ServerTestParams) *whisper.Envelope {
-	data := make([]byte, 8+whisper.TopicLength)
+	bloom := whisper.TopicToBloom(p.topic)
+	data := make([]byte, 8)
 	binary.BigEndian.PutUint32(data, p.low)
 	binary.BigEndian.PutUint32(data[4:], p.upp)
-	copy(data[8:], p.topic[:])
+	data = append(data, bloom...)
 
 	key, err := shh.GetSymKey(keyID)
 	if err != nil {

whisper/whisperv6/doc.go

@@ -60,7 +60,7 @@ const (
 	aesKeyLength    = 32 // in bytes
 	aesNonceLength  = 12 // in bytes; for more info please see cipher.gcmStandardNonceSize & aesgcm.NonceSize()
 	keyIDSize       = 32 // in bytes
-	bloomFilterSize = 64 // in bytes
+	BloomFilterSize = 64 // in bytes
 	flagsLength     = 1
 
 	EnvelopeHeaderLength = 20

whisper/whisperv6/envelope.go

@@ -208,6 +208,10 @@ func (e *Envelope) OpenSymmetric(key []byte) (msg *ReceivedMessage, err error) {
 
 // Open tries to decrypt an envelope, and populates the message fields in case of success.
 func (e *Envelope) Open(watcher *Filter) (msg *ReceivedMessage) {
+	if watcher == nil {
+		return nil
+	}
+
 	// The API interface forbids filters doing both symmetric and asymmetric encryption.
 	if watcher.expectsAsymmetricEncryption() && watcher.expectsSymmetricEncryption() {
 		return nil
@@ -249,7 +253,7 @@ func (e *Envelope) Bloom() []byte {
 
 // TopicToBloom converts the topic (4 bytes) to the bloom filter (64 bytes)
 func TopicToBloom(topic TopicType) []byte {
-	b := make([]byte, bloomFilterSize)
+	b := make([]byte, BloomFilterSize)
 	var index [3]int
 	for j := 0; j < 3; j++ {
 		index[j] = int(topic[j])

whisper/whisperv6/filter.go

@@ -35,6 +35,7 @@ type Filter struct {
 	PoW        float64           // Proof of work as described in the Whisper spec
 	AllowP2P   bool              // Indicates whether this filter is interested in direct peer-to-peer messages
 	SymKeyHash common.Hash       // The Keccak256Hash of the symmetric key, needed for optimization
+	id         string            // unique identifier
 
 	Messages map[common.Hash]*ReceivedMessage
 	mutex    sync.RWMutex
@@ -43,6 +44,10 @@ type Filter struct {
 // Filters represents a collection of filters
 type Filters struct {
 	watchers map[string]*Filter
+
+	topicMatcher     map[TopicType]map[*Filter]struct{} // map a topic to the filters that are interested in being notified when a message matches that topic
+	allTopicsMatcher map[*Filter]struct{}               // list all the filters that will be notified of a new message, no matter what its topic is
+
 	whisper *Whisper
 	mutex   sync.RWMutex
 }
@@ -51,6 +56,8 @@ type Filters struct {
 func NewFilters(w *Whisper) *Filters {
 	return &Filters{
 		watchers:         make(map[string]*Filter),
+		topicMatcher:     make(map[TopicType]map[*Filter]struct{}),
+		allTopicsMatcher: make(map[*Filter]struct{}),
 		whisper:          w,
 	}
 }
@@ -81,7 +88,9 @@ func (fs *Filters) Install(watcher *Filter) (string, error) {
 		watcher.SymKeyHash = crypto.Keccak256Hash(watcher.KeySym)
 	}
 
+	watcher.id = id
 	fs.watchers[id] = watcher
+	fs.addTopicMatcher(watcher)
 	return id, err
 }
 
@@ -91,12 +100,51 @@ func (fs *Filters) Uninstall(id string) bool {
 	fs.mutex.Lock()
 	defer fs.mutex.Unlock()
 	if fs.watchers[id] != nil {
+		fs.removeFromTopicMatchers(fs.watchers[id])
 		delete(fs.watchers, id)
 		return true
 	}
 	return false
 }
 
+// addTopicMatcher adds a filter to the topic matchers.
+// If the filter's Topics array is empty, it will be tried on every topic.
+// Otherwise, it will be tried on the topics specified.
+func (fs *Filters) addTopicMatcher(watcher *Filter) {
+	if len(watcher.Topics) == 0 {
+		fs.allTopicsMatcher[watcher] = struct{}{}
+	} else {
+		for _, t := range watcher.Topics {
+			topic := BytesToTopic(t)
+			if fs.topicMatcher[topic] == nil {
+				fs.topicMatcher[topic] = make(map[*Filter]struct{})
+			}
+			fs.topicMatcher[topic][watcher] = struct{}{}
+		}
+	}
+}
+
+// removeFromTopicMatchers removes a filter from the topic matchers
+func (fs *Filters) removeFromTopicMatchers(watcher *Filter) {
+	delete(fs.allTopicsMatcher, watcher)
+	for _, topic := range watcher.Topics {
+		delete(fs.topicMatcher[BytesToTopic(topic)], watcher)
+	}
+}
+
+// getWatchersByTopic returns a slice containing the filters that
+// match a specific topic
+func (fs *Filters) getWatchersByTopic(topic TopicType) []*Filter {
+	res := make([]*Filter, 0, len(fs.allTopicsMatcher))
+	for watcher := range fs.allTopicsMatcher {
+		res = append(res, watcher)
+	}
+	for watcher := range fs.topicMatcher[topic] {
+		res = append(res, watcher)
+	}
+	return res
+}
+
 // Get returns a filter from the collection with a specific ID
 func (fs *Filters) Get(id string) *Filter {
 	fs.mutex.RLock()
@@ -112,11 +160,10 @@ func (fs *Filters) NotifyWatchers(env *Envelope, p2pMessage bool) {
 	fs.mutex.RLock()
 	defer fs.mutex.RUnlock()
 
-	i := -1 // only used for logging info
-	for _, watcher := range fs.watchers {
-		i++
+	candidates := fs.getWatchersByTopic(env.Topic)
+	for _, watcher := range candidates {
 		if p2pMessage && !watcher.AllowP2P {
-			log.Trace(fmt.Sprintf("msg [%x], filter [%d]: p2p messages are not allowed", env.Hash(), i))
+			log.Trace(fmt.Sprintf("msg [%x], filter [%s]: p2p messages are not allowed", env.Hash(), watcher.id))
 			continue
 		}
 
@@ -128,10 +175,10 @@ func (fs *Filters) NotifyWatchers(env *Envelope, p2pMessage bool) {
 			if match {
 				msg = env.Open(watcher)
 				if msg == nil {
-					log.Trace("processing message: failed to open", "message", env.Hash().Hex(), "filter", i)
+					log.Trace("processing message: failed to open", "message", env.Hash().Hex(), "filter", watcher.id)
 				}
 			} else {
-				log.Trace("processing message: does not match", "message", env.Hash().Hex(), "filter", i)
+				log.Trace("processing message: does not match", "message", env.Hash().Hex(), "filter", watcher.id)
 			}
 		}
 
@@ -144,20 +191,6 @@ func (fs *Filters) NotifyWatchers(env *Envelope, p2pMessage bool) {
 	}
 }
 
-func (f *Filter) processEnvelope(env *Envelope) *ReceivedMessage {
-	if f.MatchEnvelope(env) {
-		msg := env.Open(f)
-		if msg != nil {
-			return msg
-		}
-
-		log.Trace("processing envelope: failed to open", "hash", env.Hash().Hex())
-	} else {
-		log.Trace("processing envelope: does not match", "hash", env.Hash().Hex())
-	}
-	return nil
-}
-
 func (f *Filter) expectsAsymmetricEncryption() bool {
 	return f.KeyAsym != nil
 }
@@ -194,16 +227,17 @@ func (f *Filter) Retrieve() (all []*ReceivedMessage) {
 
 // MatchMessage checks if the filter matches an already decrypted
 // message (i.e. a Message that has already been handled by
-// MatchEnvelope when checked by a previous filter)
+// MatchEnvelope when checked by a previous filter).
+// Topics are not checked here, since this is done by topic matchers.
 func (f *Filter) MatchMessage(msg *ReceivedMessage) bool {
 	if f.PoW > 0 && msg.PoW < f.PoW {
 		return false
 	}
 
 	if f.expectsAsymmetricEncryption() && msg.isAsymmetricEncryption() {
-		return IsPubKeyEqual(&f.KeyAsym.PublicKey, msg.Dst) && f.MatchTopic(msg.Topic)
+		return IsPubKeyEqual(&f.KeyAsym.PublicKey, msg.Dst)
 	} else if f.expectsSymmetricEncryption() && msg.isSymmetricEncryption() {
-		return f.SymKeyHash == msg.SymKeyHash && f.MatchTopic(msg.Topic)
+		return f.SymKeyHash == msg.SymKeyHash
 	}
 	return false
 }
@@ -211,27 +245,9 @@ func (f *Filter) MatchMessage(msg *ReceivedMessage) bool {
 // MatchEnvelope checks if it's worth decrypting the message. If
 // it returns `true`, client code is expected to attempt decrypting
 // the message and subsequently call MatchMessage.
+// Topics are not checked here, since this is done by topic matchers.
 func (f *Filter) MatchEnvelope(envelope *Envelope) bool {
-	if f.PoW > 0 && envelope.pow < f.PoW {
-		return false
-	}
-
-	return f.MatchTopic(envelope.Topic)
-}
-
-// MatchTopic checks that the filter captures a given topic.
-func (f *Filter) MatchTopic(topic TopicType) bool {
-	if len(f.Topics) == 0 {
-		// any topic matches
-		return true
-	}
-
-	for _, bt := range f.Topics {
-		if matchSingleTopic(topic, bt) {
-			return true
-		}
-	}
-	return false
+	return f.PoW <= 0 || envelope.pow >= f.PoW
 }
 
 func matchSingleTopic(topic TopicType, bt []byte) bool {

whisper/whisperv6/filter_test.go

@@ -303,9 +303,8 @@ func TestMatchEnvelope(t *testing.T) {
 		t.Fatalf("failed generateMessageParams with seed %d: %s.", seed, err)
 	}
 
-	params.Topic[0] = 0xFF // ensure mismatch
+	params.Topic[0] = 0xFF // topic mismatch
 
-	// mismatch with pseudo-random data
 	msg, err := NewSentMessage(params)
 	if err != nil {
 		t.Fatalf("failed to create new message with seed %d: %s.", seed, err)
@@ -314,14 +313,6 @@ func TestMatchEnvelope(t *testing.T) {
 	if err != nil {
 		t.Fatalf("failed Wrap with seed %d: %s.", seed, err)
 	}
-	match := fsym.MatchEnvelope(env)
-	if match {
-		t.Fatalf("failed MatchEnvelope symmetric with seed %d.", seed)
-	}
-	match = fasym.MatchEnvelope(env)
-	if match {
-		t.Fatalf("failed MatchEnvelope asymmetric with seed %d.", seed)
-	}
 
 	// encrypt symmetrically
 	i := mrand.Int() % 4
@@ -337,7 +328,7 @@ func TestMatchEnvelope(t *testing.T) {
 	}
 
 	// symmetric + matching topic: match
-	match = fsym.MatchEnvelope(env)
+	match := fsym.MatchEnvelope(env)
 	if !match {
 		t.Fatalf("failed MatchEnvelope() symmetric with seed %d.", seed)
 	}
@@ -396,7 +387,7 @@ func TestMatchEnvelope(t *testing.T) {
 	// asymmetric + matching topic: match
 	fasym.Topics[i] = fasym.Topics[i+1]
 	match = fasym.MatchEnvelope(env)
-	if match {
+	if !match {
 		t.Fatalf("failed MatchEnvelope(asymmetric + matching topic) with seed %d.", seed)
 	}
 
@@ -431,7 +422,8 @@ func TestMatchEnvelope(t *testing.T) {
 	// filter with topic + envelope without topic: mismatch
 	fasym.Topics = fsym.Topics
 	match = fasym.MatchEnvelope(env)
-	if match {
+	if !match {
+		// topic mismatch should have no affect, as topics are handled by topic matchers
 		t.Fatalf("failed MatchEnvelope(filter without topic + envelope without topic) with seed %d.", seed)
 	}
 }
@@ -487,7 +479,8 @@ func TestMatchMessageSym(t *testing.T) {
 
 	// topic mismatch
 	f.Topics[index][0]++
-	if f.MatchMessage(msg) {
+	if !f.MatchMessage(msg) {
+		// topic mismatch should have no affect, as topics are handled by topic matchers
 		t.Fatalf("failed MatchEnvelope(topic mismatch) with seed %d.", seed)
 	}
 	f.Topics[index][0]--
@@ -580,7 +573,8 @@ func TestMatchMessageAsym(t *testing.T) {
 
 	// topic mismatch
 	f.Topics[index][0]++
-	if f.MatchMessage(msg) {
+	if !f.MatchMessage(msg) {
+		// topic mismatch should have no affect, as topics are handled by topic matchers
 		t.Fatalf("failed MatchEnvelope(topic mismatch) with seed %d.", seed)
 	}
 	f.Topics[index][0]--
@@ -829,8 +823,9 @@ func TestVariableTopics(t *testing.T) {
 
 		f.Topics[i][lastTopicByte]++
 		match = f.MatchEnvelope(env)
-		if match {
-			t.Fatalf("MatchEnvelope symmetric with seed %d, step %d: false positive.", seed, i)
+		if !match {
+			// topic mismatch should have no affect, as topics are handled by topic matchers
+			t.Fatalf("MatchEnvelope symmetric with seed %d, step %d.", seed, i)
 		}
 	}
 }

whisper/whisperv6/peer.go

@@ -56,7 +56,7 @@ func newPeer(host *Whisper, remote *p2p.Peer, rw p2p.MsgReadWriter) *Peer {
 		powRequirement: 0.0,
 		known:          set.New(),
 		quit:           make(chan struct{}),
-		bloomFilter:    makeFullNodeBloom(),
+		bloomFilter:    MakeFullNodeBloom(),
 		fullNode:       true,
 	}
 }
@@ -120,7 +120,7 @@ func (peer *Peer) handshake() error {
 		err = s.Decode(&bloom)
 		if err == nil {
 			sz := len(bloom)
-			if sz != bloomFilterSize && sz != 0 {
+			if sz != BloomFilterSize && sz != 0 {
 				return fmt.Errorf("peer [%x] sent bad status message: wrong bloom filter size %d", peer.ID(), sz)
 			}
 			peer.setBloomFilter(bloom)
@@ -229,7 +229,7 @@ func (peer *Peer) notifyAboutBloomFilterChange(bloom []byte) error {
 func (peer *Peer) bloomMatch(env *Envelope) bool {
 	peer.bloomMu.Lock()
 	defer peer.bloomMu.Unlock()
-	return peer.fullNode || bloomFilterMatch(peer.bloomFilter, env.Bloom())
+	return peer.fullNode || BloomFilterMatch(peer.bloomFilter, env.Bloom())
 }
 
 func (peer *Peer) setBloomFilter(bloom []byte) {
@@ -238,13 +238,13 @@ func (peer *Peer) setBloomFilter(bloom []byte) {
 	peer.bloomFilter = bloom
 	peer.fullNode = isFullNode(bloom)
 	if peer.fullNode && peer.bloomFilter == nil {
-		peer.bloomFilter = makeFullNodeBloom()
+		peer.bloomFilter = MakeFullNodeBloom()
 	}
 }
 
-func makeFullNodeBloom() []byte {
-	bloom := make([]byte, bloomFilterSize)
-	for i := 0; i < bloomFilterSize; i++ {
+func MakeFullNodeBloom() []byte {
+	bloom := make([]byte, BloomFilterSize)
+	for i := 0; i < BloomFilterSize; i++ {
 		bloom[i] = 0xFF
 	}
 	return bloom

whisper/whisperv6/peer_test.go

@@ -23,6 +23,7 @@ import (
 	mrand "math/rand"
 	"net"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -71,7 +72,7 @@ var keys = []string{
 }
 
 type TestData struct {
-	started int
+	started int64
 	counter [NumNodes]int
 	mutex   sync.RWMutex
 }
@@ -151,7 +152,7 @@ func resetParams(t *testing.T) {
 }
 
 func initBloom(t *testing.T) {
-	masterBloomFilter = make([]byte, bloomFilterSize)
+	masterBloomFilter = make([]byte, BloomFilterSize)
 	_, err := mrand.Read(masterBloomFilter)
 	if err != nil {
 		t.Fatalf("rand failed: %s.", err)
@@ -163,7 +164,7 @@ func initBloom(t *testing.T) {
 		masterBloomFilter[i] = 0xFF
 	}
 
-	if !bloomFilterMatch(masterBloomFilter, msgBloom) {
+	if !BloomFilterMatch(masterBloomFilter, msgBloom) {
 		t.Fatalf("bloom mismatch on initBloom.")
 	}
 }
@@ -177,7 +178,7 @@ func initialize(t *testing.T) {
 
 	for i := 0; i < NumNodes; i++ {
 		var node TestNode
-		b := make([]byte, bloomFilterSize)
+		b := make([]byte, BloomFilterSize)
 		copy(b, masterBloomFilter)
 		node.shh = New(&DefaultConfig)
 		node.shh.SetMinimumPoW(masterPow)
@@ -240,9 +241,7 @@ func startServer(t *testing.T, s *p2p.Server) {
 		t.Fatalf("failed to start the fisrt server.")
 	}
 
-	result.mutex.Lock()
-	defer result.mutex.Unlock()
-	result.started++
+	atomic.AddInt64(&result.started, 1)
 }
 
 func stopServers() {
@@ -472,7 +471,10 @@ func checkPowExchange(t *testing.T) {
 func checkBloomFilterExchangeOnce(t *testing.T, mustPass bool) bool {
 	for i, node := range nodes {
 		for peer := range node.shh.peers {
-			if !bytes.Equal(peer.bloomFilter, masterBloomFilter) {
+			peer.bloomMu.Lock()
+			equals := bytes.Equal(peer.bloomFilter, masterBloomFilter)
+			peer.bloomMu.Unlock()
+			if !equals {
 				if mustPass {
 					t.Fatalf("node %d: failed to exchange bloom filter requirement in round %d. \n%x expected \n%x got",
 						i, round, masterBloomFilter, peer.bloomFilter)
@@ -500,11 +502,13 @@ func checkBloomFilterExchange(t *testing.T) {
 
 func waitForServersToStart(t *testing.T) {
 	const iterations = 200
+	var started int64
 	for j := 0; j < iterations; j++ {
 		time.Sleep(50 * time.Millisecond)
-		if result.started == NumNodes {
+		started = atomic.LoadInt64(&result.started)
+		if started == NumNodes {
 			return
 		}
 	}
-	t.Fatalf("Failed to start all the servers, running: %d", result.started)
+	t.Fatalf("Failed to start all the servers, running: %d", started)
 }

whisper/whisperv6/whisper.go

@@ -232,11 +232,11 @@ func (whisper *Whisper) SetMaxMessageSize(size uint32) error {
 
 // SetBloomFilter sets the new bloom filter
 func (whisper *Whisper) SetBloomFilter(bloom []byte) error {
-	if len(bloom) != bloomFilterSize {
+	if len(bloom) != BloomFilterSize {
 		return fmt.Errorf("invalid bloom filter size: %d", len(bloom))
 	}
 
-	b := make([]byte, bloomFilterSize)
+	b := make([]byte, BloomFilterSize)
 	copy(b, bloom)
 
 	whisper.settings.Store(bloomFilterIdx, b)
@@ -558,14 +558,14 @@ func (whisper *Whisper) Subscribe(f *Filter) (string, error) {
 // updateBloomFilter recalculates the new value of bloom filter,
 // and informs the peers if necessary.
 func (whisper *Whisper) updateBloomFilter(f *Filter) {
-	aggregate := make([]byte, bloomFilterSize)
+	aggregate := make([]byte, BloomFilterSize)
 	for _, t := range f.Topics {
 		top := BytesToTopic(t)
 		b := TopicToBloom(top)
 		aggregate = addBloom(aggregate, b)
 	}
 
-	if !bloomFilterMatch(whisper.BloomFilter(), aggregate) {
+	if !BloomFilterMatch(whisper.BloomFilter(), aggregate) {
 		// existing bloom filter must be updated
 		aggregate = addBloom(whisper.BloomFilter(), aggregate)
 		whisper.SetBloomFilter(aggregate)
@@ -701,7 +701,7 @@ func (whisper *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error {
 		case bloomFilterExCode:
 			var bloom []byte
 			err := packet.Decode(&bloom)
-			if err == nil && len(bloom) != bloomFilterSize {
+			if err == nil && len(bloom) != BloomFilterSize {
 				err = fmt.Errorf("wrong bloom filter size %d", len(bloom))
 			}
 
@@ -779,11 +779,11 @@ func (whisper *Whisper) add(envelope *Envelope, isP2P bool) (bool, error) {
 		}
 	}
 
-	if !bloomFilterMatch(whisper.BloomFilter(), envelope.Bloom()) {
+	if !BloomFilterMatch(whisper.BloomFilter(), envelope.Bloom()) {
 		// maybe the value was recently changed, and the peers did not adjust yet.
 		// in this case the previous value is retrieved by BloomFilterTolerance()
 		// for a short period of peer synchronization.
-		if !bloomFilterMatch(whisper.BloomFilterTolerance(), envelope.Bloom()) {
+		if !BloomFilterMatch(whisper.BloomFilterTolerance(), envelope.Bloom()) {
 			return false, fmt.Errorf("envelope does not match bloom filter, hash=[%v], bloom: \n%x \n%x \n%x",
 				envelope.Hash().Hex(), whisper.BloomFilter(), envelope.Bloom(), envelope.Topic)
 		}
@@ -928,24 +928,6 @@ func (whisper *Whisper) Envelopes() []*Envelope {
 	return all
 }
 
-// Messages iterates through all currently floating envelopes
-// and retrieves all the messages, that this filter could decrypt.
-func (whisper *Whisper) Messages(id string) []*ReceivedMessage {
-	result := make([]*ReceivedMessage, 0)
-	whisper.poolMu.RLock()
-	defer whisper.poolMu.RUnlock()
-
-	if filter := whisper.filters.Get(id); filter != nil {
-		for _, env := range whisper.envelopes {
-			msg := filter.processEnvelope(env)
-			if msg != nil {
-				result = append(result, msg)
-			}
-		}
-	}
-	return result
-}
-
 // isEnvelopeCached checks if envelope with specific hash has already been received and cached.
 func (whisper *Whisper) isEnvelopeCached(hash common.Hash) bool {
 	whisper.poolMu.Lock()
@@ -1043,12 +1025,12 @@ func isFullNode(bloom []byte) bool {
 	return true
 }
 
-func bloomFilterMatch(filter, sample []byte) bool {
+func BloomFilterMatch(filter, sample []byte) bool {
 	if filter == nil {
 		return true
 	}
 
-	for i := 0; i < bloomFilterSize; i++ {
+	for i := 0; i < BloomFilterSize; i++ {
 		f := filter[i]
 		s := sample[i]
 		if (f | s) != f {
@@ -1060,8 +1042,8 @@ func bloomFilterMatch(filter, sample []byte) bool {
 }
 
 func addBloom(a, b []byte) []byte {
-	c := make([]byte, bloomFilterSize)
-	for i := 0; i < bloomFilterSize; i++ {
+	c := make([]byte, BloomFilterSize)
+	for i := 0; i < BloomFilterSize; i++ {
 		c[i] = a[i] | b[i]
 	}
 	return c

whisper/whisperv6/whisper_test.go

@@ -75,10 +75,6 @@ func TestWhisperBasic(t *testing.T) {
 	if len(mail) != 0 {
 		t.Fatalf("failed w.Envelopes().")
 	}
-	m := w.Messages("non-existent")
-	if len(m) != 0 {
-		t.Fatalf("failed w.Messages.")
-	}
 
 	derived := pbkdf2.Key([]byte(peerID), nil, 65356, aesKeyLength, sha256.New)
 	if !validateDataIntegrity(derived, aesKeyLength) {
@@ -593,7 +589,7 @@ func TestCustomization(t *testing.T) {
 	}
 
 	// check w.messages()
-	id, err := w.Subscribe(f)
+	_, err = w.Subscribe(f)
 	if err != nil {
 		t.Fatalf("failed subscribe with seed %d: %s.", seed, err)
 	}
@@ -602,11 +598,6 @@ func TestCustomization(t *testing.T) {
 	if len(mail) > 0 {
 		t.Fatalf("received premature mail")
 	}
-
-	mail = w.Messages(id)
-	if len(mail) != 2 {
-		t.Fatalf("failed to get whisper messages")
-	}
 }
 
 func TestSymmetricSendCycle(t *testing.T) {
@@ -835,11 +826,11 @@ func TestSymmetricSendKeyMismatch(t *testing.T) {
 func TestBloom(t *testing.T) {
 	topic := TopicType{0, 0, 255, 6}
 	b := TopicToBloom(topic)
-	x := make([]byte, bloomFilterSize)
+	x := make([]byte, BloomFilterSize)
 	x[0] = byte(1)
 	x[32] = byte(1)
-	x[bloomFilterSize-1] = byte(128)
-	if !bloomFilterMatch(x, b) || !bloomFilterMatch(b, x) {
+	x[BloomFilterSize-1] = byte(128)
+	if !BloomFilterMatch(x, b) || !BloomFilterMatch(b, x) {
 		t.Fatalf("bloom filter does not match the mask")
 	}
 
@@ -851,11 +842,11 @@ func TestBloom(t *testing.T) {
 	if err != nil {
 		t.Fatalf("math rand error")
 	}
-	if !bloomFilterMatch(b, b) {
+	if !BloomFilterMatch(b, b) {
 		t.Fatalf("bloom filter does not match self")
 	}
 	x = addBloom(x, b)
-	if !bloomFilterMatch(x, b) {
+	if !BloomFilterMatch(x, b) {
 		t.Fatalf("bloom filter does not match combined bloom")
 	}
 	if !isFullNode(nil) {
@@ -865,16 +856,16 @@ func TestBloom(t *testing.T) {
 	if isFullNode(x) {
 		t.Fatalf("isFullNode false positive")
 	}
-	for i := 0; i < bloomFilterSize; i++ {
+	for i := 0; i < BloomFilterSize; i++ {
 		b[i] = byte(255)
 	}
 	if !isFullNode(b) {
 		t.Fatalf("isFullNode false negative")
 	}
-	if bloomFilterMatch(x, b) {
+	if BloomFilterMatch(x, b) {
 		t.Fatalf("bloomFilterMatch false positive")
 	}
-	if !bloomFilterMatch(b, x) {
+	if !BloomFilterMatch(b, x) {
 		t.Fatalf("bloomFilterMatch false negative")
 	}
 
@@ -888,7 +879,7 @@ func TestBloom(t *testing.T) {
 		t.Fatalf("failed to set bloom filter: %s", err)
 	}
 	f = w.BloomFilter()
-	if !bloomFilterMatch(f, x) || !bloomFilterMatch(x, f) {
+	if !BloomFilterMatch(f, x) || !BloomFilterMatch(x, f) {
 		t.Fatalf("retireved wrong bloom filter")
 	}
 }