From e47820d164e104865f334530f79ab13938dbe04f Mon Sep 17 00:00:00 2001 From: Vlad Date: Wed, 7 Dec 2016 15:28:40 +0100 Subject: [PATCH] whisper: refactored message processing --- whisper/shhapi/api.go | 16 ++++++++++ whisper/shhapi/api_test.go | 9 ++++++ whisper/whisperv5/filter.go | 4 +++ whisper/whisperv5/peer_test.go | 2 ++ whisper/whisperv5/whisper.go | 55 ++++++++++++++++++++++++++-------- 5 files changed, 74 insertions(+), 12 deletions(-) diff --git a/whisper/shhapi/api.go b/whisper/shhapi/api.go index 6ed3e17c24..f2597e133e 100644 --- a/whisper/shhapi/api.go +++ b/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 { diff --git a/whisper/shhapi/api_test.go b/whisper/shhapi/api_test.go index 13a7cee668..a10e2e4760 100644 --- a/whisper/shhapi/api_test.go +++ b/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 { diff --git a/whisper/whisperv5/filter.go b/whisper/whisperv5/filter.go index fd5f5083f6..091a31cc9e 100644 --- a/whisper/whisperv5/filter.go +++ b/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 { diff --git a/whisper/whisperv5/peer_test.go b/whisper/whisperv5/peer_test.go index 082e7f446c..88da59bff9 100644 --- a/whisper/whisperv5/peer_test.go +++ b/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() } } diff --git a/whisper/whisperv5/whisper.go b/whisper/whisperv5/whisper.go index dc9571f6ea..f8a170930d 100644 --- a/whisper/whisperv5/whisper.go +++ b/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) + } } }