From 8e720f2123e0f211befafef6e2fc24546dc0e35c Mon Sep 17 00:00:00 2001 From: Vlad Date: Mon, 12 Dec 2016 19:20:33 +0100 Subject: [PATCH] whisper: refactoring of message queuing --- whisper/whisperv5/doc.go | 10 ++++----- whisper/whisperv5/filter.go | 4 ++-- whisper/whisperv5/filter_test.go | 8 +++---- whisper/whisperv5/whisper.go | 37 ++++++++++++++++---------------- 4 files changed, 29 insertions(+), 30 deletions(-) diff --git a/whisper/whisperv5/doc.go b/whisper/whisperv5/doc.go index 8f5453e35c..e2e255e9e1 100644 --- a/whisper/whisperv5/doc.go +++ b/whisper/whisperv5/doc.go @@ -40,11 +40,11 @@ const ( ProtocolVersionStr = "5.0" ProtocolName = "shh" - statusCode = 0 - messagesCode = 1 - p2pCode = 2 - mailRequestCode = 3 - NumberOfMessageCodes = 32 + statusCode = 0 // used by whisper protocol + messagesCode = 1 // normal whisper message + p2pCode = 2 // peer-to-peer message (to be consumed by the peer, but not forwarded any futher) + p2pRequestCode = 3 // peer-to-peer message, used by Dapp protocol + NumberOfMessageCodes = 64 paddingMask = byte(3) signatureFlag = byte(4) diff --git a/whisper/whisperv5/filter.go b/whisper/whisperv5/filter.go index 99cfa54650..1cd1181fe5 100644 --- a/whisper/whisperv5/filter.go +++ b/whisper/whisperv5/filter.go @@ -76,11 +76,11 @@ func (fs *Filters) Get(i uint32) *Filter { return fs.watchers[i] } -func (fs *Filters) NotifyWatchers(env *Envelope, messageCode uint64) { +func (fs *Filters) NotifyWatchers(env *Envelope, p2pMessage bool) { fs.mutex.RLock() var msg *ReceivedMessage for j, watcher := range fs.watchers { - if messageCode == p2pCode && !watcher.AcceptP2P { + if p2pMessage && !watcher.AcceptP2P { glog.V(9).Infof("msg [%x], filter [%d]: p2p messages are not allowed \n", env.Hash(), j) continue } diff --git a/whisper/whisperv5/filter_test.go b/whisper/whisperv5/filter_test.go index bfa45782a0..9f2a58f28f 100644 --- a/whisper/whisperv5/filter_test.go +++ b/whisper/whisperv5/filter_test.go @@ -540,7 +540,7 @@ func TestWatchers(t *testing.T) { } for i = 0; i < NumMessages; i++ { - filters.NotifyWatchers(envelopes[i], messagesCode) + filters.NotifyWatchers(envelopes[i], false) } var total int @@ -593,7 +593,7 @@ func TestWatchers(t *testing.T) { } for i = 0; i < NumMessages; i++ { - filters.NotifyWatchers(envelopes[i], messagesCode) + filters.NotifyWatchers(envelopes[i], false) } for i = 0; i < NumFilters; i++ { @@ -629,7 +629,7 @@ func TestWatchers(t *testing.T) { // test AcceptP2P total = 0 - filters.NotifyWatchers(envelopes[0], p2pCode) + filters.NotifyWatchers(envelopes[0], true) for i = 0; i < NumFilters; i++ { mail = tst[i].f.Retrieve() @@ -646,7 +646,7 @@ func TestWatchers(t *testing.T) { } f.AcceptP2P = true total = 0 - filters.NotifyWatchers(envelopes[0], p2pCode) + filters.NotifyWatchers(envelopes[0], true) for i = 0; i < NumFilters; i++ { mail = tst[i].f.Retrieve() diff --git a/whisper/whisperv5/whisper.go b/whisper/whisperv5/whisper.go index 9a01083714..e673ff921c 100644 --- a/whisper/whisperv5/whisper.go +++ b/whisper/whisperv5/whisper.go @@ -36,11 +36,6 @@ 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 { @@ -61,7 +56,8 @@ type Whisper struct { mailServer MailServer - messageQueue chan QueueItem + messageQueue chan *Envelope + p2pMsgQueue chan *Envelope quit chan struct{} overflow bool @@ -79,7 +75,8 @@ func NewWhisper(server MailServer) *Whisper { expirations: make(map[uint32]*set.SetNonTS), peers: make(map[*Peer]struct{}), mailServer: server, - messageQueue: make(chan QueueItem, messageQueueLimit), + messageQueue: make(chan *Envelope, messageQueueLimit), + p2pMsgQueue: make(chan *Envelope, messageQueueLimit), quit: make(chan struct{}), } whisper.filters = NewFilters(whisper) @@ -134,7 +131,7 @@ func (w *Whisper) RequestHistoricMessages(peerID []byte, data []byte) error { return err } p.trusted = true - return p2p.Send(p.ws, mailRequestCode, data) + return p2p.Send(p.ws, p2pRequestCode, data) } func (w *Whisper) SendP2PMessage(peerID []byte, envelope *Envelope) error { @@ -369,7 +366,7 @@ func (wh *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error { wh.postEvent(envelope, p2pCode) } } - case mailRequestCode: + case p2pRequestCode: // Must be processed if mail server is implemented. Otherwise ignore. if wh.mailServer != nil { s := rlp.NewStream(packet.Payload, uint64(packet.Size)) @@ -467,9 +464,12 @@ func (w *Whisper) postEvent(envelope *Envelope, messageCode uint64) { // currently supported version, we can not decrypt it, // and therefore just ignore this message if envelope.Ver() <= EnvelopeVersion { - w.checkOverflow() - i := QueueItem{hash: envelope.Hash(), code: messageCode} - w.messageQueue <- i + if messagesCode == messageCode { + w.checkOverflow() + w.messageQueue <- envelope + } else { + w.p2pMsgQueue <- envelope + } } } @@ -491,18 +491,17 @@ func (w *Whisper) checkOverflow() { // processQueue delivers the messages to the watchers during the lifetime of the whisper node. func (w *Whisper) processQueue() { + var e *Envelope for { select { case <-w.quit: return - case i := <-w.messageQueue: - w.poolMu.Lock() - e := w.envelopes[i.hash] - w.poolMu.Unlock() - if e != nil { - w.filters.NotifyWatchers(e, i.code) - } + case e = <-w.messageQueue: + w.filters.NotifyWatchers(e, false) + + case e = <-w.p2pMsgQueue: + w.filters.NotifyWatchers(e, true) } } }