[rabbitmq-discuss] Multi-threaded Dispatch in Java Client

Tue Oct 5 19:32:29 BST 2010

Darren,

That repo is out of date. All work on this bug moved to hg.rabbitmq.com.

In the latest patch all consumers get the new threading behaviour and the JavaDoc is updated to reflect that.

I'm pretty concerned about sending callback methods to consumers on different threads. As a bare minimum we have to ensure that handleDelivery is never called out of order. 

Not just that but I think we need to ensure that exit from handleDelivery happens-before each subsequent entry into handleDelivery on a given consumer. 

This prevents us from using a rudimentary producer/consumer model. Such a model would allow for multiple hD calls for a given consumer to run concurrently.

Beyond just a single dispatch, I'm happy with a model that maintains a pool of threads but assigns consumers to one thread or another.

I'm concerned about the impact on user code of allowing dispatches to a consumer in multiple threads.

Rob
On 2 Oct 2010, at 21:55, darr wrote:

> 
> http://bitbucket.org/robharrop/rabbitmq-java-client/changeset/ea463bbd4871
> That's the commit I'm referencing.  I hope it's the right one.
> 
> It looks like you're breaking the contract of Consumer a little bit when you
> move this off of the Connection's thread (since that's what's documented). 
> This is a Good Thing, because, as you note, implementations of consumer can
> now call blocking methods on Connection and Channel.  However, this has a
> down side in that only some consumers are permitted to call blocking methods
> -- those that use DispatchingConsumer.  Other consumers are guaranteed to
> have their methods called from the Connection thread.
> 
> So this is a bit of extra complexity for implementations of Consumer to
> consider.  Given that implementors of Consumer already need to think of
> these things, you might re-evaluate the idea that consumers require all of
> their methods to be called from a single thread.  The mechanics of making a
> consumer thread-safe seem very manageable.  There's not a lot of interaction
> between the methods and very little state.
> 
> However, if you share my concern about breaking the contract of Consumer,
> you could introduce another interface (ConcurrentConsumer) which specifies
> exactly the requirements of its implementations and modify
> DispachingConsumer to accept only ConcurrentConsumer.
> 
> As for the possibility of ConnectionFactory.setDispatchThreadCount(int), I
> think if I set that to some value greater than one, and had only a single
> consumer, I would expect handleDelivery to be called from multiple threads. 
> I think we might be safer making DispatchingConsumer the sole party
> responsible for these things and it can accept an Executor on construction.
> Then the user can decide if all consumers can safely share the same set of
> threads, or if they need some other behavior.
> 
> What do you think?
> 
> Thanks,
> Darren.
> 
> 
> Rob Harrop-5 wrote:
>> 
>> Recently I pushed branch bug18384 that changes the way callbacks are sent
>> to Consumer implementations.
>> 
>> Following this change, the Connection maintains a dispatch thread that is
>> used to send callbacks to the Consumers. This frees Consumers call
>> blocking methods on the Connection and Channel.
>> 
>> A question came up on Twitter about making this configurable, allowing for
>> a custom Executor to be plugged into the ConnectionFactory. I wanted to
>> outline why this is complicated, discuss a possible implementation and see
>> if there is much interest.
>> 
>> First off, we should establish that each Consumer should only receive
>> callbacks in a single thread. If this is not the case, then chaos will
>> ensue and Consumers will need to worry about their own thread safety
>> beyond that of initialisation safety.
>> 
>> With only a single dispatch thread for all Consumers, this Consumer-Thread
>> pairing is easy honoured.
>> 
>> When we introduce multiple threads, we have to ensure that each Consumer
>> is paired with only one thread. When using the Executor abstraction, this
>> prevents each callback dispatch from being wrapped up in a Runnable and
>> sent to Executor, because you cannot guarantee which thread will be used.
>> 
>> To get around this, the Executor can be set to run 'n' long-running tasks
>> (n being the number of threads in the Executor). Each of these tasks pulls
>> dispatch instructions off a queue and executes them. Each Consumer is
>> paired with one dispatch instruction queue, probably assigned on a
>> round-robin basis. This is not too complicated and will provide simple
>> balancing of dispatch load across the threads in the Executor.
>> 
>> Now, there are still some problems:
>> 
>> 1. The number of threads in an Executor is not necessarily fixed (as with
>> ThreadPoolExecutor).
>> 2. There is no way, via Executor or ExecutorService to find out how many
>> threads there are. Thus, we cannot know how many dispatch instruction
>> queues to create.
>> 
>> However, we can certainly introduce a
>> ConnectionFactory.setDispatchThreadCount(int). Behind the scenes this will
>> create an Executors.newFixedThreadPool() and the correct number of
>> dispatch queues and dispatch tasks.
>> 
>> I'm interested in hearing if anyone thinks I'm overlooking some simpler
>> way of solving this, and indeed if this is even worth solving.
>> 
>> Regards,
>> 
>> Rob
>> _______________________________________________
>> rabbitmq-discuss mailing list
>> rabbitmq-discuss at lists.rabbitmq.com
>> https://lists.rabbitmq.com/cgi-bin/mailman/listinfo/rabbitmq-discuss
>> 
>> 
> 
> -- 
> View this message in context: http://old.nabble.com/Multi-threaded-Dispatch-in-Java-Client-tp29860178p29868134.html
> Sent from the RabbitMQ mailing list archive at Nabble.com.
> 
> _______________________________________________
> rabbitmq-discuss mailing list
> rabbitmq-discuss at lists.rabbitmq.com
> https://lists.rabbitmq.com/cgi-bin/mailman/listinfo/rabbitmq-discuss

-------------- next part --------------
An HTML attachment was scrubbed...
URL: <http://lists.rabbitmq.com/pipermail/rabbitmq-discuss/attachments/20101005/c164711d/attachment-0001.htm>