How is the API zero copy?

[RajivKurian]

From the wiki:
private static final UnsafeBuffer BUFFER = new UnsafeBuffer(ByteBuffer.allocateDirect(256));

...
final String message = "Hello World!";
BUFFER.putBytes(0, message.getBytes());

final boolean result = publication.offer(BUFFER, 0, message.getBytes().length);

Here one is allocating a buffer, and then the publication is presumably appending the bytes to the log buffer. From the StrangeLoop presentation it seemed like one would move the tail of the log buffer with a LOCK XADD and scribble their data straight onto the log buffer.

[RajivKurian]

I was expecting something like this:

// Just allocate a buffer flyweight object without having it point to real memory.
private static final UnsafeBuffer BUFFER = new UnsafeBuffer(0);

// This tries to reserve lengthOfMyData bytes on the underlying logbuffer.
// If we managed to reserve the appropriate number of bytes, then change BUFFER's internals to the right pointer + length.
final boolean result = publication.offer(BUFFER, lengthOfMyData);
if (result) {
// Do optional bounds checking to make sure we don't go past lengthOfMyData.
BUFFER.putByte(byte1);
BUFFER.putByte(byte2);
.....
}

[mjpt777]

The statement is that the data is copy-free in the processing the log buffers. This is very unusual in a message transport.

However it is funny that you mention because this week I'm working on an additional API that will allow the claiming of a range in the log buffer so it can be written directly into. This will not really matter when going off box but should make a difference in the IPC case on box. The API will look something like:

final LogClaim logClaim = new LogClaim(); // Can be stored for future use

if (publication.claim(messageLength, logClaim))
{
    final MutableDirectBuffer buffer = logClaim.buffer();
    final int offset = logClaim.offset();

    // insert data directly into the buffer or wrap with a flyweight

    logClaim.publish();
}

[RajivKurian]

Awesome!

[mjpt777]

I've pushed a commit that offers this functionality. 8ed70c4

Works like the following:

final BufferClaim bufferClaim = new BufferClaim(); // Can be stored for future use

if (publication.tryClaim(messageLength, bufferClaim))
{
    final MutableDirectBuffer buffer = bufferClaim .buffer();
    final int offset = bufferClaim .offset();

    // insert data directly into the buffer or wrap with a flyweight

    bufferClaim .commit();
}

[mikeb01]

What will happen to the state of the LogBuffer if client code calls tryClaim(), but never calls BufferClaim.commit(). Would it be useful to support passing in a lambda and ensure that the commit is always called?

public boolean tryClaim(final int length, final BufferClaim bufferClaim, Consumer<BufferClaim> callback)
{
    boolean claimed = tryClaim(length, bufferClaim);
    if (claimed)
    {
        try
        {
            callback.accept(bufferClaim);
        }
        finally
        {
            bufferClaim.commit();
        }
    }

    return claimed;
}

[mjpt777]

It is a good point and something people have to be carefully about. The issue with such a lambda is that is it likely to be capturing and thus will allocate and have an impact on latency.

[mikeb01]

You can work around the capturing, by allowing arguments to be floated through. The Disruptor's EventSink interface does this. See the description about using Java 8, it has a couple of notes around capturing Lambdas.

[mjpt777]

In your example bb will be captured and thus allocation will take place.

[mikeb01]

Not in the second example, where a method reference is used.

[mjpt777]

OK I see what you are doing. This can work for a single argument which might be sufficient for many cases. Could be an option consider.

[mikeb01]

I supported up to three and varargs (although varargs will potentially allocate). In the multi-producer case, failing to call commit/publish will leave the Disruptor (and similarly the LogBuffer - I think) in an invalid and unrepairable state. The most common case of this occurring is when the caller fails to do their exception handling correctly (commit must be in a finally block). With the Disruptor I recommend the publishEvent call as the default approach only only use next()/publish() if some advanced use case is required. I would suggest something similar for Aeron.

[RichardWarburton]

Varargs definitely allocates. Its also probably worse than a capturing lambda as well, because it needs to initialise the array of arguments.

I think really the only people who should be using this API are advanced users of the IPC case who are being bottlenecked by the additional copying cost of the current API. Do we want to caveat that in the javadoc for this class?

[mjpt777]

This is even more advanced that the Disruptor as you get direct access to the underlying buffer. Definitely an advanced feature only.