rewroks impl to use unsafe unicast sink with MpScQueue #2892
Conversation
| Sinks.Many<String> sink = Sinks.unsafe() | ||
| .many() | ||
| .unicast() | ||
| .onBackpressureBuffer(new MpscUnboundedArrayQueue<>(256)); |
There was a problem hiding this comment.
this should resolve problems with high CPU contention since that queue is designed for Multiple producers scenarios
There was a problem hiding this comment.
Alternatively, we can have a bounded Multiproducer queue here since as far as I understand we can drop part of the values
| .onBackpressureLatest(); | ||
| } else { | ||
| publisher = this.sink.asFlux(); | ||
| publisher = this.sink.asFlux().publish().autoConnect(); |
There was a problem hiding this comment.
actually - this may be optional. I don't know if we have tests for that but if we do I will try to remove that and put only in cases where we have retry In place
| sink = new LogbackMetricsSuppressingManySink(sink); | ||
| } | ||
|
|
||
| this.sink = sink; |
There was a problem hiding this comment.
Updating this.sink here does not update the reference that already registered meters have, which is why some of the StatsdMeterRegistryPublishTest tests are failing. For example, when the registry is stopped and then started, this.sink will be a new sink, but an already created StatsdCounter will still reference the old sink.
closes #2880