[ADDED] Conn.Barrier() API #338
Conversation
Suppose that you have a connection with two subscriptions, let's call them "pub" and "close". Let's say that you know that when receiving a message on "close", no new message will arrive on "pub" (because the sender does send "pub" and then a final message on "close"). But because "pub" and "close" are two different subscriptions, there is no guarantee that the message on "close" is not going to be dispatched before the last message on "pub". The Barrier() API can be used to solve this issue. In the example above, if the message callback on "close" were to call that API, this will ensure that all pending messages on "pub" have been dipatched before the provided function in the Barrier() API gets executed. See `test/TestBarrier()` test for more details.
There was a problem hiding this comment.
Generally looks good, a few comments. Question, could I not do this using channels where I could add those subscriptions such that they share a channel and a Go routine? Or set of Go routines depending on how strict you want the ordering as per invoke a function vs complete it or guarantee they are last.
| nc.subsMu.Lock() | ||
| defer nc.subsMu.Unlock() | ||
| // Need to figure out how many non chan subscriptions there are | ||
| numSubs := 0 |
There was a problem hiding this comment.
For small numbers should be fine, but should we consider tracking a variable on nc? nc.numAsyncSubs
There was a problem hiding this comment.
I'd rather leave this for now, avoiding any issue with discrepancies between +1 and -1 when removing a subscription.
| barrier := &barrierInfo{refs: int64(numSubs), f: f} | ||
| for _, sub := range nc.subs { | ||
| sub.mu.Lock() | ||
| if sub.mch == nil { |
There was a problem hiding this comment.
This should be consistent with above when counting, but if its not, does f() never get called or called too early? Should we care?
There was a problem hiding this comment.
If barrier is address, could we just count in place here and add it to the barrier struct after this iteration?
There was a problem hiding this comment.
comment 1: the nc.subsMutex is held for the duration of this call so no subscription can be removed after counting it.
comment 2: no because we still do individual sub locking, so once we post the barrier message, it is possible that a subscription processes it (after releasing the sub.mu lock here) before we have post them all. The dispatcher will possibly decrement below 0 or fire too soon.
There was a problem hiding this comment.
Maybe I misunderstood comment 1. I do not count sync and chan subscriptions because we don't have real control on when messages get dequeued (for sure for chan subscription, I would have no place to check for that special message).
|
@derekcollison Not sure if I follow, but if the idea would be to use single go channel to use for several |
|
You could queue the closed over f() with message etc. and dispatch to a group of Go routines. |
|
LGTM |
Suppose that you have a connection with two subscriptions, let's
call them "pub" and "close".
Let's say that you know that when receiving a message on "close",
no new message will arrive on "pub" (because the sender does
send "pub" and then a final message on "close"). But because
"pub" and "close" are two different subscriptions, there is no
guarantee that the message on "close" is not going to be dispatched
before the last message on "pub".
The Barrier() API can be used to solve this issue. In the example
above, if the message callback on "close" were to call that API,
this will ensure that all pending messages on "pub" have been
dipatched before the provided function in the Barrier() API gets
executed.
See
test/TestBarrier()test for more details.