[manager/dispatcher] Synchronize Dispatcher.Stop() with incoming rpcs. #2519
Conversation
anshulpundir
force-pushed
the
dispatcher
branch
2 times, most recently
from
68bf7b5
to
788dd37
Compare
Codecov Report
@@ Coverage Diff @@
## master #2519 +/- ##
==========================================
+ Coverage 61.3% 61.31% +0.01%
==========================================
Files 133 133
Lines 21710 21724 +14
==========================================
+ Hits 13309 13320 +11
- Misses 6948 6962 +14
+ Partials 1453 1442 -11 |
anshulpundir
force-pushed
the
dispatcher
branch
from
788dd37
to
38d3fee
Compare
|
why not combine rpcRW and mu? |
manager/dispatcher/dispatcher.go
Outdated
| d.rpcRW.RLock() | ||
| defer d.rpcRW.RUnlock() | ||
|
|
||
| if !d.isRunning() { |
There was a problem hiding this comment.
This is needed to fail Heartbeat() calls after the d.nodes has been cleared. Should be OK to call without a lock because dispatcher is stopped (basically wanted to prevent grabbing a lock on every Heartbeat)
cc @fcrisciani
There was a problem hiding this comment.
@anshulpundir thanks for clarifying. This also needs to be a comment here.
I have similar sentiments, but trying to keep it simple for now. Also need to get this in by tomorrow @dperny |
manager/dispatcher/dispatcher.go
Outdated
| // According to go's documentation on WaitGroup, | ||
| // Add() with a positive delta that occur when the counter is zero | ||
| // must happen before a Wait(). | ||
| // As is, dispatcher's top() can race with Run(). | ||
| d.wg.Wait() |
There was a problem hiding this comment.
@anshulpundir what is this waitgroup currently required for?
There was a problem hiding this comment.
Its used in case Stop() is called before the dispatcher Run has a chance to initialize the dispatcher.
anshulpundir
force-pushed
the
dispatcher
branch
2 times, most recently
from
e694fe2
to
0b5ac34
Compare
There was a problem hiding this comment.
Overall structure LGTM.
@anshulpundir I'm okay with combining mu and rpcRW in a follow-up PR, but we should probably do that soon.
I have not done an exhaustive analysis on if its actually a good idea to replace mu with rpcRW, or if it actually reduces complexity rather than add to it. So, whether we should do it will probably depend on that. LMK if you are convinced that it should happen and I can definitely prioritize it! |
|
LGTM |
anshulpundir
force-pushed
the
dispatcher
branch
from
0b5ac34
to
32aadc5
Compare
| // According to go's documentation on WaitGroup, | ||
| // Add() with a positive delta that occur when the counter is zero | ||
| // must happen before a Wait(). | ||
| // As is, dispatcher Stop() can race with Run(). | ||
| d.wg.Wait() |
There was a problem hiding this comment.
wait, what does this waitgroup do now?
There was a problem hiding this comment.
oh i hadn't refreshed before i commented
There was a problem hiding this comment.
Signed-off-by: Anshul Pundir <anshul.pundir@docker.com>
anshulpundir
force-pushed
the
dispatcher
branch
from
32aadc5
to
6b163ff
Compare
|
LGTM, will merge when CI goes green. |
Fixes for synchronizing the dispatcher shutdown with in-progress rpcs. This addresses the case where the Dispatcher.Register() rpc races with Dispatcher.Stop() and reinserts into the dispatcher node store after it has been cleaned up by Stop().