WithWaitUntilQueued option for topic.Publish #232
Conversation
aarshkshah1992
force-pushed
the
feat/Publish-WaitUntilQueued
branch
from
f8a2f96
to
46a7ac1
Compare
topic.go
Outdated
| // WithWaitUntilQueued blocks the publish until the message has been added to the outbound message queue for | ||
| // as many peers as the arg indicates | ||
| // A value of -1 means all peers in mesh/fanout for gossipsub & all subscribed peers in floodsub | ||
| // Please note that if nPeers is -1, the behavior is not fail fast |
There was a problem hiding this comment.
Does a "non fail fast" behaviour make sense here ? If the caller is willing to wait for the message to be added to the queues for all peers, I figured we might as well wait till the end so we can log all failures for better debugging later.
| 0, | ||
| -1, | ||
| nil, | ||
| }, |
There was a problem hiding this comment.
What's the best way to test for dropped messages ? As in, "Failure for the Wait for all peers" scenario.
| if hasListener { | ||
| select { | ||
| case <-listener.ctx.Done(): | ||
| case listener.notifChan <- &msgQueuedNotification{pid, success}: |
There was a problem hiding this comment.
This is way too heavy, it is going to spam with events for every message sent.
Also note that the channel could block the event loop, which is a no,no.
There was a problem hiding this comment.
I see what you mean.
We could use a buffered channel here & drop the event if the buffer is full.
But, that could lead to lost events & Publish would end up erroneously reporting that it wasn't able to fulfil the target.
There was a problem hiding this comment.
So yeah, this seems quite impossible to implement correctly.
We can't block the event loop and we don't want to erroneously report non-delivery (or not report at all).
There was a problem hiding this comment.
@raulk @aschmahmann Wdyt ? Can we use a buffered channel here & do the WaitUntilQueued on a "best effort" basis ?
Or, is there a smarter way to do this ?
There was a problem hiding this comment.
we could launch a goroutine for every channel put, which would side-step the problem. But I worry about potentially unbounded number of goroutines running if the receiver is stalling.
There was a problem hiding this comment.
@vyzo @aschmahmann Aha, here is one thing that can work:
For WaitUntilNQueued:
Buffer size = N& the event loop ONLY writes success cases to the channel- Event loop drops messages if the buffer is full
For WaitiUntillAllQueued:
This is slightly tricky as I've explained here. But, we can do it like so:
Buffer size = len(pubsub.topics[topic])(number of peers interested in the topic) when Publish is called- Event loop writes both success & failure cases to the channel
- Event loop spins up a new go-routine if the buffer is full. Since we have estimated the buffer size based on the number of peers we know are interested in the topic when Publish is called, it effectively prevents spinning up too many go-routines.
Let me know what you think :)
There was a problem hiding this comment.
Hey @aschmahmann @vyzo
Please take a look at this when you can.
|
@vyzo I understand your concern. Please can you give some suggestions on how to approach this ? |
|
Let me think about it. |
|
Also, apologies for jumping the gun on this one and implementing it too soon. Just wanted to get my hands dirty with pubsub. Will be careful going ahead. |
|
No need to apologize! Your enthusiasm is very much welcome! |
|
Closing as we need to revisit the semantics of this whole thing. |
Implementation for the
WaitUntilQueuedoption as specified in #217 & enlisted in #220.@aschmahmann @vyzo @Stebalien
Please take a look. This is my first "meaty" PubSub PR & I'd really love to know your feedback/if there is a better way to do this.