ARTEMIS-2945 Artemis native JNI code can be replaced by Java

[franz1981]

https://issues.apache.org/jira/browse/ARTEMIS-2945

[franz1981]

I've noticed a thing I would like to raise here to be sure I'll remember in the future :)

I see in the original code that on submitWrite we used to do:

// The GlobalRef will be deleted when poll is called. this is done so// the vm wouldn't crash if the Callback passed by the user is GCed between submission// and callback.// also as the real intention is to hold the reference until the life cycle is completeiocb->data = (void *) (*env)->NewGlobalRef(env, callback);

But I see that isn't happening for jobject bufferWrite (ie the direct ByteBuffer used to perform I/O); AFAIK it shouldn't be a problem for 2 reasons:

1. the Artemis code hold a ByteBuffer reference into IOSequentialCallback::buffer until the callback is being called (done or onError).
2. such buffers are being allocated using JNI NewDirectByteBuffer on newAlignedBuffer with the comment

// This will allocate a buffer, aligned by alignment.// Buffers created here need to be manually destroyed by destroyBuffer, or this would leak on the process heap away of Java's GC managed memory// NOTE: this buffer will contain non initialized data, you must fill it up properly

Hence I can simplify a bit the java code here, because the ByteBuffer content used while submitting I/O won't ever be GCed while being used by the kernel.

[franz1981]

@clebertsuconic@michaelandrepearce
I've used the chance that writing Java code simplify changing and trying new logics and I've introduced a new mechanism to batch fdatasyncs while preserving some fairness to read operation.

Let's explain what master does first: on master the LibaioContext blocked poll logic batches completed writes on the same file descriptor altogether; given that fdatasync is blocking, it allows to pile up new incoming write requests in batches, favouring a bursty behaviour. This come for free by not processing the already completed writes, blocked by fdatasync: this would create backpressure both on kernel level (because the completion ring buffer head isn't moved forward) and both on Java side, because the SemaphoreioSpace used to backpressure write/read submissions would run out of permissions if the existing in-flight ones are not processed (ie completed and fdatasync'd).

This nice mechanism raise a legit question:

• what happen to a read operations that has been correctly submitted?

Assuming that reads are rare on Artemis, they can still happen, during compaction, and a slow read means that compaction would suffer as well: on master a storm of writes could easily prevent a completed read to be processed because the blocked poll loop is busy processing fdatasync, that can take "long" arbitrary time.

There are several ways to fix this issue and TBH the first commit of this same PR already take care of this, but assume a smart submitter of writes/reads, that correctly batch writes and reads on user space (ie Java land - the artemis TimedBuffer).
Given that master instead is relying on a backpressuring mechanism to allow a misconfigured TimedBuffer to perform well, I've tried to batch fdatasync allowing 1 pending asynchronous fdatasync each time while fixing the latency of read, because the blocked poll event loop won't block on a long synchronous blocking operation anymore.

Not sure is the right path but is a nice way to show how easy is to bring in new behaviours :)

[franz1981]

Just FYI I've run some tests on a not recent hardware and I still get the best performance with lower number of producers/consumers (without compaction in the mix) from
using the original logic (but written in java) ie

intlastFile = -1;
for (inti = 0; i < events; i++) {
finalIoEventArray.IoEventioEvent = ioEventArray.get(i);
assertioEvent.obj() != 0;
finalintid = (int) ioEvent.data();
finalPooledIOCBpooledIOCB = iocbArray[id];
assertioEvent.obj() == pooledIOCB.address;
assertIoCb.aioData(pooledIOCB.bytes) == id;
finalSubmitInfosubmitInfo = pooledIOCB.submitInfo;
if (submitInfo != null) {
pooledIOCB.submitInfo = null;
}
finallongres = ioEvent.res();
if (res >= 0) {
finalintfd = IoCb.aioFildes(pooledIOCB.bytes);
if (fd != dumbFD) {
if (useFdatasync) {
if (lastFile != fd) {
lastFile = fd;
fdatasync(fd);
}
}
} else {
stop = true;
}
}
iocbPool.add(pooledIOCB);
if (ioSpace != null) {
ioSpace.release();
}
if (submitInfo != null) {
if (res >= 0) {
submitInfo.done();
} else {
// TODO the error string can be cached?submitInfo.onError((int) -res, strError(res));
}
}
}

On higher number of producers/consumers both the commits on this PR (ie not batching async fdsync and batching async fdsync) perform better then the original logic (in C or Java).

[franz1981]

@clebertsuconic@michaelandrepearce

I've just noticed that thanks to the AioRing abstraction we could create an AioRing.CompletionCallback interface and use it to be truly zero-copy, decoding on the fly IoEvents into their primitive fields (data, obj, res, res2) while looping trought completions: with the simplified code on #9 (comment) this should be very simple to be implemented and effective, given that a long fdatasync would likely let some more completions to accumulate while looping.

This is what I mean:
https://github.com/franz1981/activemq-artemis-native-1/blob/58696f37472eb722e6bd0af760b2f2beaa4e2a2b/src/main/java/org/apache/activemq/artemis/nativo/jlibaio/AioRing.java#L107-L145

publicinterfaceAioRingCompletionCallback {
voidhandle(longdata, longobj, longres, longres2);
}
publicintpoll(AioRingCompletionCallbackcallback, intmin, intmax) {
finallongnrAddress = this.nrAddress;
finallongheadAddress = this.headAddress;
finallongtailAddress = this.tailAddress;
finallongioEventsAddress = this.ioEventsAddress;
finalintnr = UNSAFE.getInt(nrAddress);
inthead = UNSAFE.getInt(headAddress);
// no need of membar here because Unsafe::getInt already provide itfinalinttail = UNSAFE.getIntVolatile(null, tailAddress);
intavailable = tail - head;
if (available < 0) {
// a wrap has occurredavailable += nr;
}
if (available < min) {
return0;
}
if (available == 0) {
return0;
}
// this is to mitigate a RHEL BUG: see native code for more infoif (available > nr) {
return -1;
}
available = Math.min(available, max);
for (inti = 0; i < available; i++) {
finallongioEvent = ioEventsAddress + (head * SIZE_OF_IO_EVENT_STRUCT);
finallongdata = UNSAFE.getLong(ioEvent + IoEventArray.IoEvent.DATA_OFFSET);
finallongobj = UNSAFE.getLong(ioEvent + IoEventArray.IoEvent.OBJ_OFFSET);
finallongres = UNSAFE.getLong(ioEvent + IoEventArray.IoEvent.RES_OFFSET);
finallongres2 = UNSAFE.getLong(ioEvent + IoEventArray.IoEvent.RES2_OFFSET);
head++;
head = head >= nr ? 0 : head;
UNSAFE.putOrderedInt(null, headAddress, head);
callback.handle(data, obj, res, res2);
}
returnavailable;
}

[franz1981]

I'm going to close this and will later create a bug fix for the read fairness instead: #10 is a more straight one that just try to replace what we already have instead