Ring buffer application

[Rajiv Kurian]

On a network thread I receive byte buffers which are really arrays of objects, where each object is composed of 3 longs. I need to send these objects in a sharded manner to a couple threads for processing. There are multiple threads for parallelism only. I can't send the entire bytebuffer in one operation since different parts of it will be processed by different parallel threads based on some form of hashing. I am thinking of doing one of the following for my use case:

1) Use a ring buffer of wrapper objects. My objects will just have a reference to a bytebuffer and an index into it. When a bytebuffer comes in on the network, my network thread will change the objects on the ring buffer to point to the bytebuffer and set the index appropriately. The consumer will process the payload(slices of the original bytebuffer) and set the bytebuffer reference to null. So when all slices get processed the bytebuffer should get GCed. Here I don't copy any of the bytebuffer contents. I have multiple references floating around, can't just reuse the original bytebuffer and might possibly incur the wrath of the GC gods. I also have some false sharing since I need to set the bytebuffer refs to null on the consumer.

2) Use a ring buffer of objects that have 3 longs. When a bytebuffer comes in, I can read it and copy the long fields onto ring buffer objects. The bytebuffer should get GCed or become available for reuse as soon as all the copying is done. I pay with copying but it's just 3 longs per object as opposed to a reference and an int in (1).

3) I create a specialized primitive friendly version of the ring buffer where I use a bytebuffer for my objects encoded in a way similar to the bytebuffer I receive on the network. When I get an incoming request i check to see find contiguous runs of data that needs to be sent to a single thread. I can then just do the memcpy equivalent of Java to copy from source to destination (might require Unsafe) and set the writer sequence appropriately. This is really like (2) except it is feeding into a single slab of memory instead of going through objects.

Of course benchmarking would be the right way to find out the best alternative. Any comments from experienced users/maintainers on the merits of each approach?

[Michael Barker]

I would probably do a variation of 2&3.  If you data is definition just a 3 longs I would create a custom ring buffer consisting of a single array of longs which would be 3 * ring buffer size and pack the 3 longs next to each other in the array.  I'd use a ByteBuffer for reading from the network and recycle it on the networking thread.  I suspect you'd probably only need one ByteBuffer per thread.  To get an efficient copy on the long array you could do:

ByteBuffer bb = ByteBuffer.allocateDirect(...);

// do read

LongBuffer lb = bb.asLongBuffer();

long[] ring = ...

lb.get(ring, sequence * 3, 3);

This given an efficient memory copy without using Unsafe.

Mike.

--
You received this message because you are subscribed to the Google Groups "Disruptor" group.
To unsubscribe from this group and stop receiving emails from it, send an email to lmax-disrupto...@googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

[Rajiv Kurian]

Thanks Mike! I actually have a few more types in my object - an extra byte which contains some flags including how to interpret one of the 64bit fields(either a long or a double) - hence I thought a bytebuffer would be a good fit.

I am hoping I can use some parts of the disruptor, especially the padded sequences since it seems like you guys have found a reliable way of separating the producer and consumer sequences onto separate cache lines.

[Michael Barker]

Hi Rajiv,

In this talk (http://yow.eventer.com/yow-2013-1080/disruptor-3-0-details-and-advanced-patterns-by-michael-barker-1381) I discuss how to separate the concurrent sequencing code from the underlying data store.  I include an example using a backing ByteBuffer (https://github.com/LMAX-Exchange/disruptor/blob/master/src/perftest/java/com/lmax/disruptor/offheap/OneToOneOffHeapThroughputTest.java).  You may find this useful.

Mike.

[Rajiv Kurian]

Awesome. Will see the talk and look at the example.

[Nitsan Wakart]

You are in luck ;-)
I've had the great help (as in he did all the actual work) of Richard W. in creating an off heap ring buffer with code generated flyweight object nicely hiding the Unsafe underbelly. It's in development and still experimental but the syntax we're aiming for is:
// somewhere:
SpscChannel<Example> channel = new SpscChannel<>(buffer, CAPACITY, Example.class);
// Producer thread:
ChannelProducer<Example> producer = channel.producer();
if(producer.claim()){
Example writer = producer.getWriter();
writer.setFoo(5);
writer.setBar(10L);
producer.commit();
}
// Consumer thread:
// receiverFunction implements receive(Example e);
consumer = channel.consumer(receiverFunction);
consumer.read();

Example is an interface with get/set Foo/Bar currently all primitives.
Have a look in JCTools(https://github.com/JCTools/JCTools) experimental subproject, we should be benchmarking/finalizing soon.
This is an SPSC unicast, not the disruptor multicast pattern. It's an off heap ring buffer, so should work as an IPC mechanism. The generated code handles the packaging. The ring buffer is using the FastFlow style of SPSC.
We hope to expand the functionality and the scope of available patterns/uses in the next few weeks.
I would appreciate your feedback on the implementation/usecase with the understanding that this is very early development.

[Rajiv Kurian]

Sounds good! I'll definitely take a look. The pains one has to go to to NOT use C/C++.