[rabbitmq-discuss] Per-Connection Flow Control - RMQ 2.8.1

Thu Mar 29 20:39:40 BST 2012

Hi Simon,

I have recompiled OTP R15B and RabbitMQ-Server 2.8.1 from source.
I have two versions of 2.8.1 compiled, one with the patch below(2.8.1b) 
and one without(2.8.1a).
I need to dive into the code a little more, but hopefully there will be 
some way to monitor what is in the buffer/cache and what is entering the 
queue.

I'll post up some stats once I get everything configured in the cluster.
I sent some info in a previous message, let me know if that was helpful 
or not.

Thanks

On 3/28/12 8:42 AM, Simon MacMullen wrote:
> On 28/03/12 11:32, Simon MacMullen wrote:
>> I will try to reproduce based on the above, but anything else you can
>> tell me would be a great help.
>
> Hmm, I'm struggling with this.
>
> I can certainly get into a state with a combination of x-message-ttl, 
> lots of messages expiring, and enough memory pressure to push much of 
> the queue onto disc, such that the queue performance fluctuates quite 
> a bit (as it switches between accepting new messages / paging messages 
> out to disc / paging them back in / expiring them).
>
> But I can't get a message rate that works OK in this scenario with 
> flow control disabled but blocks for long periods with it enabled. If 
> the message rate is high enough for flow control to kick in, then it's 
> also high enough for memory use to go up as messages wait to enter the 
> queue if flow control is disabled. Eventually the memory alarm goes 
> off instead.
>
> So can you provide more information about your use pattern?
>
> Having said that, if you just want to disable flow control and are 
> prepared to build from source, you could apply the following patch:
>
> diff -r f0dafaca6cd1 src/rabbit_reader.erl
> --- a/src/rabbit_reader.erl    Mon Mar 26 17:12:41 2012 +0100
> +++ b/src/rabbit_reader.erl    Wed Mar 28 13:40:03 2012 +0100
> @@ -360,7 +360,7 @@
>
>  control_throttle(State = #v1{connection_state = CS,
>                               conserve_memory  = Mem}) ->
> -    case {CS, Mem orelse credit_flow:blocked()} of
> +    case {CS, Mem} of
>          {running,   true} -> State#v1{connection_state = blocking};
>          {blocking, false} -> State#v1{connection_state = running};
>          {blocked,  false} -> ok = rabbit_heartbeat:resume_monitor(
>
> But I would like to understand more about what you're seeing.
>
> Cheers, Simon
>