Consumer dispatcher improvements

[bollhals]

Proposed Changes

some simplifications / improvements around consumer dispatcher

Types of Changes

What types of changes does your code introduce to this project?
Put an x in the boxes that apply

• Bug fix (non-breaking change which fixes issue #NNNN)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause an observable behavior change in existing systems)
• Documentation improvements (corrections, new content, etc)
• Cosmetic change (whitespace, formatting, etc)

Checklist

• I have read the CONTRIBUTING.md document
• I have signed the CA (see https://cla.pivotal.io/sign/rabbitmq)
• All tests pass locally with my changes
• I have added tests that prove my fix is effective or that my feature works
• I have added necessary documentation (if appropriate)
• Any dependent changes have been merged and published in related repositories

Further Comments

Method	Mean	Error	StdDev	Ratio	Gen 0	Gen 1	Gen 2	Allocated
Publish_Hello_World	207.0 ms	4.13 ms	7.76 ms	1.00	2000.0000	-	-	10.35 MB
NEW_Publish_Hello_World	202.3 ms	4.01 ms	10.84 ms	1.00	1000.0000	-	-	4.84 MB

Before

Method	Count	Concurrency	Mean	Error	StdDev	Gen 0	Gen 1	Gen 2	Allocated	Code Size
AsyncConsumerDispatcher	1	1	1.229 us	0.2810 us	0.0154 us	0.0191	-	-	96 B	2103 B
ConcurrentConsumerDispatcher	1	1	1.187 us	0.0108 us	0.0006 us	0.0343	-	-	169 B	2103 B
AsyncConsumerDispatcher	1	2	1.370 us	0.5994 us	0.0329 us	0.0381	-	-	181 B	2103 B
ConcurrentConsumerDispatcher	1	2	1.958 us	0.3071 us	0.0168 us	0.0992	-	-	471 B	2103 B
AsyncConsumerDispatcher	30	1	6.357 us	0.5384 us	0.0295 us	0.6104	-	-	2880 B	2103 B
ConcurrentConsumerDispatcher	30	1	6.081 us	0.3841 us	0.0211 us	1.0757	-	-	5040 B	2103 B
AsyncConsumerDispatcher	30	2	8.267 us	0.3701 us	0.0203 us	0.6256	-	-	2978 B	2103 B
ConcurrentConsumerDispatcher	30	2	39.470 us	0.4128 us	0.0226 us	3.7842	-	-	17702 B	2103 B

After

Method	Count	Concurrency	Mean	Error	StdDev	Gen 0	Gen 1	Gen 2	Allocated	Code Size
AsyncConsumerDispatcher	1	1	1,075.4 ns	125.3 ns	6.87 ns	-	-	-	-	1721 B
ConsumerDispatcher	1	1	1,105.2 ns	135.1 ns	7.41 ns	-	-	-	-	1721 B
AsyncConsumerDispatcher	1	2	778.4 ns	133.4 ns	7.31 ns	0.0048	-	-	23 B	1721 B
ConsumerDispatcher	1	2	750.6 ns	500.3 ns	27.42 ns	0.0038	-	-	19 B	1721 B
AsyncConsumerDispatcher	30	1	6,517.4 ns	854.4 ns	46.83 ns	-	-	-	-	1721 B
ConsumerDispatcher	30	1	5,909.7 ns	333.6 ns	18.29 ns	-	-	-	-	1721 B
AsyncConsumerDispatcher	30	2	7,872.5 ns	1,586.4 ns	86.95 ns	-	-	-	39 B	1721 B
ConsumerDispatcher	30	2	7,872.6 ns	1,593.0 ns	87.32 ns	-	-	-	31 B	1721 B

Before

After

[bollhals]

rebased to master. this is ready for review

[bollhals]

What's left here to be merged? (#1009 should be merged first, I'll update this one afterwards)

[LamarLugli]

This line may cause the same exception as #1013 if the tag isn't in the dictionary.

[bollhals]

Yes, but question is, can this ever happen? If the server sends a tag twice or a wrong one, then causing an exception seems fine to me. (GetConsumerOrDefault on the other hand does need to care as the protocol allows scenario where this could happen.)

[LamarLugli]

in #1012, the system has tried to get a consumer that wasn't in the dictionary and it caused chaos in the client. It may have been related to a network glitch with timeouts when unsubscribing a consumer that caused the message to not be there, or to get sent in from the server multiple times. The dictionary exception kept reoccurring consistently and frequently. I haven't tracked down the root cause though, or spent much time trying to duplicate, but I have had that exception a enough times in production that it got on my radar to fix.

I don't think we should throw an exception here since it isn't handled well further up the stack. Logging the error seems fairly reasonable though, and may help track down any bugs. Then we can ignore subsequent actions, or resort to some default predictable behaviour (like defaultConsumer).

This function looks like it gets called on shutdown, so handling the error gracefully is probably not a big problem. I'm not familiar enough with the full code base though to comment on this aspect authoritatively.

[bollhals]

Thanks for the information. I overlooked that it happened on CancelOk.

Yes, for CancelOk there might be some overlapping in some strange scenarios possible. Look at.

If for whatever reason the server sends some other reply instead of CancelOk, the consumer will be removed here and later again when the actual CancelOk comes. This is the reason in my PR I removed this double removing.

But what interests me more is the failure to handle the exception more gracefully, I'll take a look at it sometime.

AFAIK on protocol error (e.g. getting a consumer tag that doesn't exist) should result in an exception, @michaelklishin is that right? Or how should this be handled?

[michaelklishin]

The active consumer state is present in channel state on both sides, so there is a race condition between what the client and server do. E.g. you can call a basic.cancel and immediately remove the tag, only to get a basic.cancel-ok with it later. Bot sides in this example do something that makes sense at first glance but the relative timing is problematic.

In other clients we remove the consumer after receiving a basic.cancel-ok. If we don't have a consumer but received a consumer operation frame for it, some clients log a warning. Gracefully doing nothing would also be acceptable to me.

[bollhals]

I've implemented it now in a ultra safe manner, in which the consumer is:

• found via the tag, if not
• DefaultConsumer is returned, if null then
• FallbackConsumer is returned, where the implementation will just log the call itself and continues.

[bollhals]

@LamarLugli Would you agree that #1012 / #1013 is still fixed with the current proposed implementation?

[michaelklishin]

This now has some conflicts. @bollhals please rebase 🙏

[bollhals]

This now has some conflicts. @bollhals please rebase 🙏

done

[bollhals]

Any final comments on this or can we merge it?
@stebet Any input on this? (PS: "Welcome back" 👍)

[bollhals]

@michaelklishin Can we finish this? I have another change coming that depends on this and there seem to be no open points to do.

[michaelklishin]

@bollhals sorry, I thought there was more to do. I can begin QA'ing this tomorrow.

[bollhals]

@bollhals sorry, I thought there was more to do. I can begin QA'ing this tomorrow.

Thanks. 👍