More of a question than an issue...how would I go about using an Agrona queue's element's contents to control consumption ?

The idea is to offer items to a queue to preserve their order, kick off async enrichment of the items and then only consume once enrichments have completed. I'd use a putOrderedObject to issue a store-store fence for the enrichments to ensure they were visible on the poll.

I suspect there is a more elegant way to approach this problem, but the extension to OneToOne queue below seems to work. Thoughts/improvements/suggestions appreciated.

public E poll(Predicate<E> predicate) {
        final Object[] buffer = this.buffer;
        final long currentHead = head;
        final long elementOffset = sequenceToBufferOffset(currentHead, capacity - 1);
        final Object e = UNSAFE.getObjectVolatile(buffer, elementOffset);
        if (null != e && predicate.test((E) e))
        {
            UNSAFE.putOrderedObject(buffer, elementOffset, null);
            UNSAFE.putOrderedLong(this, HEAD_OFFSET, currentHead + 1);
            return (E)e;
        }
        return null;
    }

FindBugs has a dedicated check for this pattern because it was employed in OpenJDK and emulated in many other places (it has already been fixed in OpenJDK). The trick which saves from emitting garbage also breaks code such as new ArrayList<>(map.entrySet()). AFAIK Java 8 has significantly improved Escape Analysis so a safe implementation should still be garbage-free in most usages.

https://github.com/real-logic/Agrona/blob/master/src/main/java/uk/co/real_logic/agrona/collections/Long2LongHashMap.java#L470

https://github.com/real-logic/Agrona/blob/master/src/main/java/uk/co/real_logic/agrona/collections/Long2ObjectHashMap.java#L693

… by eliminating setMemory during read.

The call to setMemory was used by the reader to zero the bytes of all read messages so that subsequently written variable length messages are guaranteed to be followed by a zero length record header. The reader observes this zero length record header to efficiently detect when to stop reading.

The responsibility for each message to be followed by a zero length record header is now shifted to the writer, which already knows the precise location of the end of each message record.

The reader now no longer writes a block of zeros to the record buffer during read.

The writer requires each message to be followed by a zero length record header, even when the ring buffer is full, so the total capacity is reduced by the size of a record header (8 bytes).

I was wondering if primitive support for List is going to be added, it might not be very common and you could argue that a Set could do but sometimes you are just transferring a List of IDs and you know they are unique.

I was using Agrona but because it was missing IntList support I had to switch to FastUtil v7.x though it is too big for my needs.

Looks like currently TimerWheel is effectively broken.

Test:

int TICK_PERIOD_NS = 10;
int TIMEOUT_NS = 12;
controlTimestamp = 0;

final AtomicLong firedTimestamp = new AtomicLong(-1);
final TimerWheel wheel = new TimerWheel(this::getControlTimestamp, 12, TimeUnit.NANOSECONDS, 8);
final Runnable task = () -> firedTimestamp.set(wheel.clock().nanoTime());

final TimerWheel.Timer timer1 = wheel.newTimeout(TIMEOUT_NS, TimeUnit.NANOSECONDS, task);

assertEquals("Nothing expired on 10 timestamp", 0, wheel.expireTimers()); //which is fine
controlTimestamp = 10;
assertEquals("Nothing expired on 10 timestamp", 0, wheel.expireTimers()); //mmm... 12 should be here
controlTimestamp = 20;
assertEquals("Nothing expired on 20 timestamp... it's already definitely an error", 0, wheel.expireTimers());

This test pass. So... tick is 10ns. We schedule execution on 12ns timestamp. Nothing executed on 10ns, nothing executed on 20ns. You can put any value in TICK_PERIOD_NS which % TICK_PERIOD_NS != 0.

Reason:

TimerWheel#

if (0 >= timer.remainingRounds)
{
    timer.remove();
    timer.state = TimerState.EXPIRED;
    if (now >= timer.deadline)
    {
        ++timersExpired;
        timer.task.run();
    }
}
else
{
    timer.remainingRounds--;
}

    final long calculatedIndex = deadline / tickDurationInNs;

So... we schedule Timer to tick number in the "floor" div mode and after that we usually get now <= timer.deadline() if now is around right tick time. As a result - we mark Timer as Expired and do nothing + remove it with timer.remove().

Proposals:

1. Main one. Stop double check deadline value. In any inaccuracy of external scheduler that makes a call of expireTimers() a bit earlier - Timers execution will be postponed additionally for ticksPerWheel * tickDurationNs. So deadline should be used only to determine tickIndex and for nothing more.

WheelTimer already depends on the external scheduler accuracy, because it fully depends on triggering ticks more or less on right intervals outside. So, it should be fine to check only remainingRounds on current tick.

if (0 >= timer.remainingRounds)
{
    timer.remove();
    timer.state = TimerState.EXPIRED;
    ++timersExpired;
    timer.task.run();
}
else
{
    timer.remainingRounds--;
}

2. Now we schedule to ticks by division in "floor" mode. For current implementation with deadline, scheduling would be safer in the "ceiling" mode. But, after implementing 1), we should assign the nearest tick as it should be done according to the idea of HashedWheelTimer.

//IntMath is a guava class, just to reference what do I mean
final long calculatedIndex = IntMath.divide(deadline, tickDurationInNs, RoundingMode.HALF_DOWN);

*) Another approach to fix design - stop doing currentTick++; on each expireTimers() invocation and rework tick triggering logic. So, when we trigger expireTimers() outside - we calculate how much ticks passed since previous execution using clock and expire-run all Timers in passed ticks. This is even better, because we stop to be dependent on accuracy of external scheduler not in terms of frequency, but in terms of accuracy and things like STW pauses which could cause ticks loosing.

And one more note about this code: now >= timer.deadline is fine if we work with milliseconds. But incorrect for nanoseconds because of long overflow. The only correct way to compare nanosecond timestamps is: now - timer.deadline >= 0

I'm quite sure it can be ignored for a time, but reporting it anyway:

WARNING: An illegal reflective access operation has occurred
WARNING: Illegal reflective access by org.agrona.IoUtil (file:/home/signals/green-engine-runner/scripts/aeron/repo/agrona-0.9.7.jar) to method sun.nio.ch.FileChannelImpl.map0(int,long,long)
WARNING: Please consider reporting this to the maintainers of org.agrona.IoUtil

[ /opt/Agrona ]# ./gradlew --stacktrace :clean :compileJava :processResources UP-TO-DATE :classes :generatePrimitiveExpansions Generated LongLongConsumer Generated LongArrayList Generated LongIterator Generated Long2LongHashMap Generated LongHashSet Generated LongLruCache Generated Long2ObjectCache Generated Long2ObjectHashMap :compileGeneratedJava :processGeneratedResources UP-TO-DATE :generatedClasses :jar :javadocjavadoc: error - invalid flag: -quıet

Usage: javadoc [options] [packagenames] [sourcefiles] [@files] -overview Read overview documentation from HTML file -public Show only public classes and members -protected Show protected/public classes and members (default) -package Show package/protected/public classes and members -private Show all classes and members -help Display command line options and exit -doclet Generate output via alternate doclet -docletpath Specify where to find doclet class files -sourcepath Specify where to find source files -classpath Specify where to find user class files -cp Specify where to find user class files -exclude Specify a list of packages to exclude -subpackages Specify subpackages to recursively load -breakiterator Compute first sentence with BreakIterator -bootclasspath Override location of class files loaded by the bootstrap class loader -source Provide source compatibility with specified release -extdirs Override location of installed extensions -verbose Output messages about what Javadoc is doing -locale Locale to be used, e.g. en_US or en_US_WIN -encoding Source file encoding name -quiet Do not display status messages -J Pass directly to the runtime system -X Print a synopsis of nonstandard options and exit

Provided by Standard doclet: -d Destination directory for output files -use Create class and package usage pages -version Include @version paragraphs -author Include @author paragraphs -docfilessubdirs Recursively copy doc-file subdirectories -splitindex Split index into one file per letter -windowtitle Browser window title for the documentation -doctitle Include title for the overview page -header Include header text for each page -footer Include footer text for each page -top Include top text for each page -bottom Include bottom text for each page -link Create links to javadoc output at -linkoffline Link to docs at using package list at -excludedocfilessubdir :.. Exclude any doc-files subdirectories with given name. -group :.. Group specified packages together in overview page -nocomment Suppress description and tags, generate only declarations. -nodeprecated Do not include @deprecated information -noqualifier ::... Exclude the list of qualifiers from the output. -nosince Do not include @since information -notimestamp Do not include hidden time stamp -nodeprecatedlist Do not generate deprecated list -notree Do not generate class hierarchy -noindex Do not generate index -nohelp Do not generate help link -nonavbar Do not generate navigation bar -serialwarn Generate warning about @serial tag -tag ::

see relevant TODO here: https://github.com/real-logic/Agrona/blob/master/agrona/src/main/java/org/agrona/concurrent/DynamicCompositeAgent.java#L214

Essentially, remove is threadsafe in it's list manipulation, but the old list may be iterated through by an AgentRunner which will call doWork after onClose has happened or as it is happening. A possible solution is to have a work counter (incremented by the doWork loop on execution) used to delay the close effect until a further iteration has run.

There is a bug in ManyToManyConcurrentArrayQueue, ManyToOneConcurrentArrayQueue and ManyToOneConcurrentLinkedQueue that leads to poll returning null even if there was an offer operation before in the same thread and there are no concurrent poll operations.

I believe that this bug differs from the one mentioned in documentation, because it is not connected to isEmpty or size methods and queues are known to be not empty.

The bug can be reproduced with Lincheck. The scenario that can produce non-linearizable results is the same for all mentioned queues:

Parallel part:
| offer(1): true | offer(5): true |
|                | poll():   null |

There are two parallel threads and the poll operation in the second thread returns null despite the fact that the queue definitely contains at least one element (the element that was added by the previous offer(5)).

Here is an example of incorrect execution for ManyToOneConcurrentLinkedQueue (yet to be released feature in Lincheck):

= Parallel part execution: =
|                      | offer(5)                                                                                                          |
|                      |   swapTail(@34bccdde): @60cd5f19 at ManyToOneConcurrentLinkedQueue.offer(ManyToOneConcurrentLinkedQueue.java:140) |
|                      |   switch                                                                                                          |
| offer(1): true       |                                                                                                                   |
| poll(): null         |                                                                                                                   |
|   thread is finished |                                                                                                                   |
|                      |   putOrderedObject(@60cd5f19,16,@34bccdde) at Node.nextOrdered(ManyToOneConcurrentLinkedQueue.java:46)            |
|                      |   result: true                                                                                                    |
|                      |   thread is finished                                                                                              |   

Whenever the IntHashSet/LongHashSet reaches the requested capacity, it will enter an infinite loop searching for an empty slot (confirmed for capacities of 1 and 2). I would propose the following:

1. the set should keep operating correctly when it has reached the requested capacity;
2. it should either document its unusual failure mode or change it to an exception.

At the moment the header is put at the end of the buffer, so we have:

    tailCounterIndex = capacity + RingBufferDescriptor.TAIL_COUNTER_OFFSET;
    headCounterIndex = capacity + RingBufferDescriptor.HEAD_COUNTER_OFFSET;

If the header was at the beginning of the buffer we could save ourselves all offset loads, E.g. instead of loading the tail counter like this:

load this.tailCounterIndex
load this.addressOffset
offset = addressOffset + tailCounterIndex
load offset

we could use a constant and eliminate the first load.

Hi real-logic/agrona!

This is a one-off automatically generated pull request from LGTM.com :robot:. You might have heard that we’ve integrated LGTM’s underlying CodeQL analysis engine natively into GitHub. The result is GitHub code scanning!

With LGTM fully integrated into code scanning, we are focused on improving CodeQL within the native GitHub code scanning experience. In order to take advantage of current and future improvements to our analysis capabilities, we suggest you enable code scanning on your repository. Please take a look at our blog post for more information.

This pull request enables code scanning by adding an auto-generated codeql.yml workflow file for GitHub Actions to your repository — take a look! We tested it before opening this pull request, so all should be working :heavy_check_mark:. In fact, you might already have seen some alerts appear on this pull request!

Where needed and if possible, we’ve adjusted the configuration to the needs of your particular repository. But of course, you should feel free to tweak it further! Check this page for detailed documentation.

Questions? Check out the FAQ below!

FAQ