whisper: refactored message processing

whisper/shhapi/api.go

@@ -55,6 +55,22 @@ func APIs() []rpc.API {
 	}
 }
 
+// Start starts the Whisper worker threads.
+func (api *PublicWhisperAPI) Start() error {
+	if api.whisper == nil {
+		return whisperOffLineErr
+	}
+	return api.whisper.Start(nil)
+}
+
+// Stop stops the Whisper worker threads.
+func (api *PublicWhisperAPI) Stop() error {
+	if api.whisper == nil {
+		return whisperOffLineErr
+	}
+	return api.whisper.Stop()
+}
+
 // Version returns the Whisper version this node offers.
 func (api *PublicWhisperAPI) Version() (*rpc.HexNumber, error) {
 	if api.whisper == nil {

whisper/shhapi/api_test.go

@@ -277,6 +277,9 @@ func TestIntegrationAsym(t *testing.T) {
 		t.Fatalf("failed to create API.")
 	}
 
+	api.Start()
+	defer api.Stop()
+
 	sig, err := api.NewIdentity()
 	if err != nil {
 		t.Fatalf("failed NewIdentity: %s.", err)
@@ -375,6 +378,9 @@ func TestIntegrationSym(t *testing.T) {
 		t.Fatalf("failed to create API.")
 	}
 
+	api.Start()
+	defer api.Stop()
+
 	keyname := "schluessel"
 	err := api.GenerateSymKey(keyname)
 	if err != nil {
@@ -471,6 +477,9 @@ func TestIntegrationSymWithFilter(t *testing.T) {
 		t.Fatalf("failed to create API.")
 	}
 
+	api.Start()
+	defer api.Stop()
+
 	keyname := "schluessel"
 	err := api.GenerateSymKey(keyname)
 	if err != nil {

whisper/whisperv5/filter.go

@@ -76,6 +76,10 @@ func (fs *Filters) Get(i uint32) *Filter {
 }
 
 func (fs *Filters) NotifyWatchers(env *Envelope, messageCode uint64) {
+	if env == nil {
+		return
+	}
+
 	fs.mutex.RLock()
 	var msg *ReceivedMessage
 	for _, watcher := range fs.watchers {

whisper/whisperv5/peer_test.go

@@ -118,6 +118,7 @@ func initialize(t *testing.T) {
 		var node TestNode
 		node.shh = NewWhisper(nil)
 		node.shh.test = true
+		node.shh.Start(nil)
 		topics := make([]TopicType, 0)
 		topics = append(topics, sharedTopic)
 		f := Filter{KeySym: sharedKey, Topics: topics}
@@ -166,6 +167,7 @@ func stopServers() {
 		n := nodes[i]
 		if n != nil {
 			n.shh.Unwatch(n.filerId)
+			n.shh.Stop()
 			n.server.Stop()
 		}
 	}

whisper/whisperv5/whisper.go

@@ -22,6 +22,7 @@ import (
 	crand "crypto/rand"
 	"crypto/sha256"
 	"fmt"
+	"runtime"
 	"sync"
 	"time"
 
@@ -35,6 +36,11 @@ import (
 	set "gopkg.in/fatih/set.v0"
 )
 
+type QueueItem struct {
+	hash common.Hash
+	code uint64
+}
+
 // Whisper represents a dark communication interface through the Ethereum
 // network, using its very own P2P communication layer.
 type Whisper struct {
@@ -55,7 +61,9 @@ type Whisper struct {
 
 	mailServer MailServer
 
-	quit chan struct{}
+	messageQueue chan QueueItem
+	quit         chan struct{}
+
 	test bool
 }
 
@@ -63,14 +71,15 @@ type Whisper struct {
 // Param s should be passed if you want to implement mail server, otherwise nil.
 func NewWhisper(server MailServer) *Whisper {
 	whisper := &Whisper{
-		privateKeys: make(map[string]*ecdsa.PrivateKey),
-		symKeys:     make(map[string][]byte),
-		envelopes:   make(map[common.Hash]*Envelope),
-		messages:    make(map[common.Hash]*ReceivedMessage),
-		expirations: make(map[uint32]*set.SetNonTS),
-		peers:       make(map[*Peer]struct{}),
-		mailServer:  server,
-		quit:        make(chan struct{}),
+		privateKeys:  make(map[string]*ecdsa.PrivateKey),
+		symKeys:      make(map[string][]byte),
+		envelopes:    make(map[common.Hash]*Envelope),
+		messages:     make(map[common.Hash]*ReceivedMessage),
+		expirations:  make(map[uint32]*set.SetNonTS),
+		peers:        make(map[*Peer]struct{}),
+		mailServer:   server,
+		messageQueue: make(chan QueueItem, 1024),
+		quit:         make(chan struct{}),
 	}
 	whisper.filters = NewFilters(whisper)
 
@@ -270,6 +279,12 @@ func (w *Whisper) Send(envelope *Envelope) error {
 func (w *Whisper) Start(*p2p.Server) error {
 	glog.V(logger.Info).Infoln("Whisper started")
 	go w.update()
+
+	numCPU := runtime.NumCPU()
+	for i := 0; i < numCPU; i++ {
+		go w.processQueue()
+	}
+
 	return nil
 }
 
@@ -444,14 +459,30 @@ func (wh *Whisper) add(envelope *Envelope) error {
 	return nil
 }
 
-// postEvent delivers the message to the watchers.
+// postEvent queues the message for further processing.
 func (w *Whisper) postEvent(envelope *Envelope, messageCode uint64) {
 	// if the version of incoming message is higher than
 	// currently supported version, we can not decrypt it,
 	// and therefore just ignore this message
 	if envelope.Ver() <= EnvelopeVersion {
-		// todo: review if you need an additional thread here
-		go w.filters.NotifyWatchers(envelope, messageCode)
+		i := QueueItem{hash: envelope.Hash(), code: messageCode}
+		w.messageQueue <- i
+	}
+}
+
+// processQueue delivers the messages to the watchers during the lifetime of the whisper node.
+func (w *Whisper) processQueue() {
+	for {
+		select {
+		case <-w.quit:
+			return
+
+		case i := <-w.messageQueue:
+			w.poolMu.Lock()
+			e := w.envelopes[i.hash]
+			w.poolMu.Unlock()
+			w.filters.NotifyWatchers(e, i.code)
+		}
 	}
 }