[msg] Do not Close singleton MessageProcessors when closing connections #3934
Conversation
This fixes a regression that was introduced in #3918 The shared singleton MessageProcessor was being closed when a Connection was being closed. Now a Pool interface is introduced. A MessageProcessor is fetched when a Connection is created and returned when a Connection is closed. It is up to the pool impl to decide if it should Close the MessageProcessor on each return or only when the pool is closed.
| @@ -63,27 +63,35 @@ func TestServerWithMessageFn(t *testing.T) { | |||
| s := server.NewServer("a", NewMessageHandler(SingletonMessageProcessor(p), opts), server.NewOptions()) | |||
| s.Serve(l) | |||
|
|
|||
| conn, err := net.Dial("tcp", l.Addr().String()) | |||
| conn1, err := net.Dial("tcp", l.Addr().String()) | |||
There was a problem hiding this comment.
this test fails without the fix (too many calls to MessageProcessor.Close)
src/msg/consumer/types.go
Outdated
| // in a connection. | ||
| type NewMessageProcessorFn func() MessageProcessor | ||
| // MessageProcessorPool returns MessageProcessors. | ||
| type MessageProcessorPool interface { |
There was a problem hiding this comment.
IMO a bit confusing using the name "pool" here since it isn't actually pooling anything, just controlling if the provided processor itself should get closed or not.
There was a problem hiding this comment.
well the singleton is definitely pooling. do you have another naming suggestion?
There was a problem hiding this comment.
Maybe change to Provider - also maybe remove Put and just have the provider differences be: singleton wraps a processor such that it's Close is no-op; otherwise provider gives a new processor that has a real Close.
There was a problem hiding this comment.
yea i originally had a wrapper that did nothing on Close. To me that seemed weird since it was so coupled the Handler impl. But it seems like this pool thingy is confusing, so i'll just go back to that previous impl.
src/msg/consumer/types.go
Outdated
|
|
||
| // SingletonMessageProcessor returns a MessageProcessorPool that shares the same MessageProcessor for all users. The | ||
| // MessageProcessor is closed when the pool is closed. | ||
| func SingletonMessageProcessor(mp MessageProcessor) MessageProcessorPool { |
There was a problem hiding this comment.
I don't see this referenced anywhere? Am I missing something?
There was a problem hiding this comment.
it already existed https://github.com/m3db/m3/blob/master/src/cmd/services/m3coordinator/server/m3msg/config.go#L89
The function signature just changed
Codecov Report
@@ Coverage Diff @@
## master #3934 +/- ##
======================================
Coverage 56.7% 56.7%
======================================
Files 557 557
Lines 63545 63545
======================================
Hits 36069 36069
Misses 24287 24287
Partials 3189 3189
Flags with carried forward coverage won't be shown. Click here to find out more. Continue to review full report at Codecov.
|
This fixes a regression that was introduced in #3918
The shared singleton MessageProcessor was being closed when a Connection
was being closed.
Now a Pool interface is introduced. A MessageProcessor is fetched when a
Connection is created and returned when a Connection is closed. It is up
to the pool impl to decide if it should Close the MessageProcessor on
each return or only when the pool is closed.
What this PR does / why we need it:
Fixes #
Special notes for your reviewer:
Does this PR introduce a user-facing and/or backwards incompatible change?:
Does this PR require updating code package or user-facing documentation?: