Hi Nitsan,

This is not for merge - just to open the discussion about efficient BlockingQueue implementations.

The SpscArrayQueueBlocking here is hand-crafted by inheritance of SpscArrayQueue but ultimately I would prefer to have the BlockingQueue Implementations dynamically generated on demand by the QueueFactory.

Let me know your thoughts before I push it too far.

Cheers Georges

In Februrary, I found your MpmcArrayQueue, and was very impressed with the performance and lack of generated heap objects (for offer/poll). Being interested in having a similar capability to LinkedTransferQueue (but without any heap allocation), I modified your MpmcArrayQueue to create the "MpmcTransferArrayQueue", using a similar technique that Doug Lea used in the LinkedTransferQueue with regards to park/unpark rules. Since one of the requirements for your library is that it can run on java6, I did not depend on the TransferQueue interface -- I'm not sure the best way to resolve this, perhaps to have offer/poll block instead of having transfer/take?? I'm up for whatever suggestions you may have on this topic.

Here are my handrolled benchmarks for this queue (based on your spsc benchmark code), and I hope that this queue can find a home in your tools. The bechmarks were run on an i7-4700HQ CPU @ 2.40GHz w/ 16gigs of ram, running Linux 3.13.0-49-generic, x86_64. I did not pin the process to CPUs.

The non-blocking test results are based on your MpmcArrayQueue implementation, with a small (1024) queue size and a busy-spin on a full queue.

| Blocking | LBQ | LTQ | MTAQ | | --- | --- | --- | --- | | 1x1 Threads | 3,7m op/s | 6,3m op/s | 3,6m op/s | | 2x2 Threads | 1,9m op/s | 2,4m op/s | 2,9m op/s | | 3x3 Threads | 1,4m op/s | 1,8m op/s | 2,0m op/s | | 4x4 Threads | 0,8m op/s | 1,4m op/s | 1,5m op/s |

| Non-Blocking | LBQ | LTQ | MTAQ | | --- | --- | --- | --- | | 1x1 Threads | 5,8m op/s | 8,1m op/s | 30,7m op/s | | 2x2 Threads | 1,4m op/s | 3,4m op/s | 6,4m op/s | | 3x3 Threads | 1,2m op/s | 2,3m op/s | 3,8m op/s | | 4x4 Threads | 0,8m op/s | 1,8m op/s | 2,4m op/s |

It seems that this project would an ideal place to have NonBlockingHashMap created by @cliffclick. Unfortunately original project that was hosted on SourceForge (https://sourceforge.net/projects/high-scale-lib/) is dead and has been for quite a while. Also it is buggy, see stephenc/high-scale-lib#5 for one such example.

The current version of NonBlockingHashMap and supporting classes is available in the https://github.com/h2oai/h2o-3/ project under water.nbhm package.

The problem with those classes being part of https://github.com/h2oai/h2o-3 project is that it is hard to consumer this project as library since one needs to always exclude transitive dependencies and consume full jar with other classes. Another much bigger problem is that recently NBHM classes become more and more specialized for H2O use cases. In fact it seems that it is no longer possible to use those classes standalone (see https://github.com/h2oai/h2o-3/commit/6ef6f834e67585c55ed6c8b48817749d2643607e).

Hence this feature request asks to add awesome NBHM classes to this awesome JCTools project.

/cc @cliffclick

Up-to-date copy of NonBlockingHashMap and friends. JUnits for all tests, including performance tests. No integration into maven, just the code and tests.

Reorder offerAndWakeup() to fix race. Previously the following sequence was possible:

Consumer                            Provider

poll(timeout)
                                  offerAndWakeup()
time out
                                  soRefElement(, offset,)
consume from offset
offset -> offset'
poll(timeout)
                                  soBlocked(null)
                                  unpark(consumerThread)
// Spin for ever
spinWaitForElement(, offset')

Would it be possible/feasible to batch consume the contents of an MpscLinkedQueue? I.e., have a method taking a callback which then traverses the links via plain loads until it reaches the producer's node. So instead of polling one by one, it would move the consumer pointer once and not bother nulling out every node's value.

Hi. I was playing with NonBlockingHashMap and found that removes can return one items more than once when executed concurrently. Here is the test sample:

@Test
public void testMap() throws Exception {
    final ExecutorService executor = Executors.newFixedThreadPool(32);

    final ConcurrentMap<String, String> map = new NonBlockingHashMap<>();
    final String key = UUID.randomUUID().toString();

    final int numberOfIterations = 10_000;
    final CountDownLatch latch = new CountDownLatch(numberOfIterations);

    CopyOnWriteArrayList<String> addedList = new CopyOnWriteArrayList<>();
    CopyOnWriteArrayList<String> removedList = new CopyOnWriteArrayList<>();

    for (int i = 0; i < numberOfIterations; i++) {
        if (i % 2 == 0) {
            executor.submit(() -> {
                String added = UUID.randomUUID().toString();
                map.put(key, added);

                addedList.add(added);

                latch.countDown();
            });
        } else {
            executor.submit(() -> {
                String removed = map.remove(key);

                if (removed != null) {
                    removedList.add(removed);
                }

                latch.countDown();
            });
        }
    }

    assertTrue(latch.await(5, TimeUnit.SECONDS));

    assertEquals(addedList.size(), Sets.newHashSet(addedList).size());
    assertEquals(removedList.size(), Sets.newHashSet(removedList).size());
}

My last assertEquals fails with number of actual removed items more than unique items.

The mechanics of this queue is similar on how Aeron's Publication (aka the LogBuffer) works, but using only 2 partitions and without background cleaning of the consumed elements. When hit by a single producer the offer and poll has the same semantics of the MpscArrayQueue, otherwise there is the chance that more than the configured capacity will be written (without element losses), exactly as the Aeron's LogBuffer. Benchmarks shows (throughput-wise) perf similar to MpscArrayQueue with Single Producer and a better scalability otherwise (mainly due to XADD).

I'm seeing AssertionErrors from time to time in our CI environment. Here are two stacktraces that show where error is generated:

java.lang.AssertionError
	at org.jctools.maps.NonBlockingHashMap$CHM.copy_check_and_promote(NonBlockingHashMap.java:1082)
	at org.jctools.maps.NonBlockingHashMap$CHM.help_copy_impl(NonBlockingHashMap.java:1036)
	at org.jctools.maps.NonBlockingHashMap$CHM.access$500(NonBlockingHashMap.java:796)
	at org.jctools.maps.NonBlockingHashMap.help_copy(NonBlockingHashMap.java:789)
	at org.jctools.maps.NonBlockingHashMap.putIfMatch(NonBlockingHashMap.java:701)
	at org.jctools.maps.NonBlockingHashMap.putIfMatch(NonBlockingHashMap.java:378)
	at org.jctools.maps.NonBlockingHashMap.put(NonBlockingHashMap.java:327)

and

java.lang.AssertionError
	at org.jctools.maps.NonBlockingHashMap$CHM.copy_check_and_promote(NonBlockingHashMap.java:1082)
	at org.jctools.maps.NonBlockingHashMap$CHM.help_copy_impl(NonBlockingHashMap.java:1036)
	at org.jctools.maps.NonBlockingHashMap$CHM.access$500(NonBlockingHashMap.java:796)
	at org.jctools.maps.NonBlockingHashMap.help_copy(NonBlockingHashMap.java:789)
	at org.jctools.maps.NonBlockingHashMap.putIfMatch(NonBlockingHashMap.java:701)
	at org.jctools.maps.NonBlockingHashMap.putIfMatch(NonBlockingHashMap.java:378)
	at org.jctools.maps.NonBlockingHashMap.remove(NonBlockingHashMap.java:343)

Here is the line in question NonBlockingHashMap.java#L1082:

assert (copyDone+workdone) <= oldlen;

While debugging this issue locality I saw the following values that lead to exception:

copyDone=32	
workdone=1	
oldlen=32

Unfortunate I don't have a standalone reproducer but in the test several threads are banging on the map (some add entries and some remove entries). And this test fails sometimes for adders and sometimes for removers.

The problem is only reproducible with -ea flag. Without it test runs green every time. So what is the purpose of such assertions if actual code run fine even when assertions fail? Maybe @cliffclick knows?

Recently, I decided to add Chinese comments to JCTools. When I read the source code of SpscLinkedQueue, the implementation of the offer caused me some confusion, because it may cause consumerNode newer than producerNode and cause bugs. Therefore, I wrote SpscLinkedQueueTest for verification, and an exception occurred.

The producer should publish the new producerNode first, and then ensure that it is reachable from the consumerNode.

        // final LinkedQueueNode<E> nextNode = newNode(e);
        // final LinkedQueueNode<E> preNode = lpProducerNode();
        // spProducerNode(nextNode);
        // preNode.soNext(nextNode);

        final LinkedQueueNode<E> nextNode = newNode(e);
        lpProducerNode().soNext(nextNode);
        spProducerNode(nextNode);

public class SpscLinkedQueueTest {

    private static volatile boolean stop = false;

    public static void main(String[] args) throws InterruptedException {
        SpscLinkedQueue<String> messageQueue = new SpscLinkedQueue<>();

        new Producer(messageQueue).start();
        new Consumer(messageQueue).start();

        try {
            Thread.sleep(10 * 1000);
        } finally {
            stop = true;
        }
    }

    private static class Producer extends Thread {

        final SpscLinkedQueue<String> messageQueue;

        long sequence = 0;

        Producer(SpscLinkedQueue<String> messageQueue) {
            this.messageQueue = messageQueue;
        }

        @Override
        public void run() {
            try {
                while (!stop) {
                    messageQueue.offer(Long.toString(sequence++));
                    Thread.sleep(1);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    private static class Consumer extends Thread {

        final SpscLinkedQueue<String> messageQueue;

        private Consumer(SpscLinkedQueue<String> messageQueue) {
            this.messageQueue = messageQueue;
        }

        @Override
        public void run() {
            while (!stop) {
                messageQueue.poll();

                if (!messageQueue.isEmpty()) {
                    final String e = messageQueue.poll();
                    if (null == e) {
                        throw new Error("MessageQueue.isEmpty() is false, messageQueue.poll() return null!");
                    }
                }
            }
        }

    }
}

Exception in thread "Thread-1" java.lang.Error: MessageQueue.isEmpty() is false, messageQueue.poll() return null! at com.wjybxx.fastjgame.concurrenttest.SpscLinkedQueueTest$Consumer.run(SpscLinkedQueueTest.java:84)

This is pretty much a straight up port of the existing perf_hash_test to JMH, per #108. I moved SimpleRandom out of perf_hash_test.

Result of running both on the same machine:

jctools-benchmarks git:(nbhm-jmh)% for t in 2 4 6 8; do java -jar target/microbenchmarks.jar ConcurrentMapsThroughput.rand -t $t -f 1; done

# Threads: 2 threads, will synchronize iterations
# Benchmark: org.jctools.maps.nhbm_test.jmh.ConcurrentMapsThroughput.randomGetPutRemove
# Parameters: (implementation = NonBlockingHashMap, readRatio = 50, tableSize = 100000)

Benchmark                                      (implementation)  (readRatio)  (tableSize)   Mode  Cnt         Score         Error  Units
ConcurrentMapsThroughput.randomGetPutRemove  NonBlockingHashMap           50       100000  thrpt    6  17923933.571 ±  912895.637  ops/s
ConcurrentMapsThroughput.randomGetPutRemove   ConcurrentHashMap           50       100000  thrpt    6  23668440.241 ± 5684267.690  ops/s

###

# Threads: 4 threads, will synchronize iterations
# Benchmark: org.jctools.maps.nhbm_test.jmh.ConcurrentMapsThroughput.randomGetPutRemove
# Parameters: (implementation = NonBlockingHashMap, readRatio = 50, tableSize = 100000)

Benchmark                                      (implementation)  (readRatio)  (tableSize)   Mode  Cnt         Score         Error  Units
ConcurrentMapsThroughput.randomGetPutRemove  NonBlockingHashMap           50       100000  thrpt    6  44864770.568 ± 1202113.075  ops/s
ConcurrentMapsThroughput.randomGetPutRemove   ConcurrentHashMap           50       100000  thrpt    6  43945366.601 ± 3180110.185  ops/s

###

# Threads: 6 threads, will synchronize iterations
# Benchmark: org.jctools.maps.nhbm_test.jmh.ConcurrentMapsThroughput.randomGetPutRemove
# Parameters: (implementation = NonBlockingHashMap, readRatio = 50, tableSize = 100000)

Benchmark                                      (implementation)  (readRatio)  (tableSize)   Mode  Cnt         Score         Error  Units
ConcurrentMapsThroughput.randomGetPutRemove  NonBlockingHashMap           50       100000  thrpt    6  57554032.876 ±  335786.149  ops/s
ConcurrentMapsThroughput.randomGetPutRemove   ConcurrentHashMap           50       100000  thrpt    6  56251688.387 ± 3140116.465  ops/s

###

# Threads: 8 threads, will synchronize iterations
# Benchmark: org.jctools.maps.nhbm_test.jmh.ConcurrentMapsThroughput.randomGetPutRemove
# Parameters: (implementation = NonBlockingHashMap, readRatio = 50, tableSize = 100000)

Benchmark                                      (implementation)  (readRatio)  (tableSize)   Mode  Cnt         Score         Error  Units
ConcurrentMapsThroughput.randomGetPutRemove  NonBlockingHashMap           50       100000  thrpt    6  64094875.527 ± 2180806.338  ops/s
ConcurrentMapsThroughput.randomGetPutRemove   ConcurrentHashMap           50       100000  thrpt    6  62433923.436 ± 2661512.313  ops/s


jctools-benchmarks git:(nbhm-jmh)% java -cp ... org.jctools.maps.nhbm_test.perf_hash_test
50% gets, 25% inserts, 25% removes, table_size=100000 Best
Threads from 2 to 8 by 2
Warmup -variance:
===     CHM_16    2  cnts/sec=   22514871
===  NBHashMap    2  cnts/sec=   16893822 size=1048578
==== HashMap  Threads   Trial:    0          1          2          3          4          5          6          Avg      Stddev
===     CHM_16    2  cnts/sec=   17159725   24732722   24932298   25377222   23633198   25029976   25464857   24898332 (+/- 0%)  65377
===  NBHashMap    2  cnts/sec=   17281654   18891777   17008697   17564187   19901123   19047840   19239117   18501268 (+/- 4%)  65841 size=1048578
===     CHM_16    4  cnts/sec=   45679183   45948000   45803727   45934142   45873051   45820472   45274057   45967458 (+/- 0%)  65452
===  NBHashMap    4  cnts/sec=   45759998   46562757   46239087   46287162   46085904   45991481   46322006   46204051 (+/- 0%)  65443 size=1048578
===     CHM_16    6  cnts/sec=   57408141   58173032   58174207   58186218   58161588   58219358   58191284   58177819 (+/- 0%)  65507
===  NBHashMap    6  cnts/sec=   57648986   58911525   59107223   58633843   58790711   58970249   58758884   58820373 (+/- 0%)  65471 size=1048578
===     CHM_16    8  cnts/sec=   64764567   64139703   64566574   64774410   65508813   65302646   65318703   64947207 (+/- 0%)  65630
===  NBHashMap    8  cnts/sec=   65516278   63649737   65449784   66149245   65772432   65584908   65647868   65583018 (+/- 0%)  65458 size=1048578

Bonus: now supports HashMap as a baseline (this is not enabled by default). If run in a single- threaded mode, the HashMap will not be synchronized.

Until now jctools-core used the Apache Felix Maven Bundle Plugin to generate its OSGI descriptor.

This PR:

• moves the internal classes to the org.jctools.util.internal. If needed this package can be exported to other JCTools artifacts using the @ExportTo annotation from BND.
• exports only explicitly annotated packages (i.e. all except org.jctools.util.internal). The OSGI descriptor becomes:

Export-Package: 
 org.jctools.queues.atomic;version="4.0.2";uses:="org.jctools.queues",
 org.jctools.queues.unpadded;version="4.0.2";uses:="org.jctools.queues",
 org.jctools.queues;version="4.0.2",
 org.jctools.maps;version="4.0.2",
 org.jctools.util;version="4.0.2",
 org.jctools.counters;version="4.0.2"
Import-Package: sun.misc;resolution:=optional
Require-Capability: osgi.ee;filter:="(&(osgi.ee=JavaSE)(version=1.8))"

• automatically generates a module-info.class JPMS descriptor, which is in sync (cf. JPMS libraries) with the OSGI descriptor:

module [email protected] {
  requires java.base;
  requires static jdk.unsupported;
  exports org.jctools.queues;
  exports org.jctools.queues.atomic;
  exports org.jctools.queues.unpadded;
  exports org.jctools.maps;
  exports org.jctools.util;
  exports org.jctools.counters;
}

This descriptor can be generated by JDK 8.

Feature Request

Add a method to NonBlockingHashMap similar to the parallel forEach method in ConcurrentHashMap.

Background

In Bazel we currently use Guava's interners, but we are looking at other strategies for interning in order to reduce storage cost. Our estimate is that > 6% of Bazel's memory cost is from the interning data structures (not including the elements, just the storage).

We are interested in switching from ConcurrentHashMap to NonBlockingHashMap primarily for the getk() method, which would allow Bazel's main graph map to double as an interner for its keys, replacing Guava interners and the storage cost they entail. A parallel iteration method would close the gap vs our current usage of ConcurrentHashMap.

Hi JCTools/JCTools!

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

Version 4.0.1 on Maven Central does not contain an Automatic-Module-Name (introduced in #289) in the manifest. I am not able to reproduce the problem compiling the v4.0.1 tag from sources, but it must be caused by the maven-jar-plugin configuration on the host that built the released artifact.

Remark that the currently used maven-bundle-plugin version supports the automatic generation of JPMS module descriptors (even on JDK 8) through the <_jpms-module-info/> instruction. The generated module descriptor is functionally equivalent to the generated OSGI bundle descriptor. I can provide a PR that modularizes jctools-core, jctools-channels and jctools-experimental.

since jdk19 has released，any plans on fitting the new virtual threads model? I noticed many implementation depends on keeping the thread spinning, might not so efficient?

Sometimes projects may need to embed part of JCTool's and can benefit by reusing some of the test infrastructure. This is a use at your own risk type of option, but is very low effort to offer with that expected caveat. By following the Maven's Guide to using attached tests, an additional artifact is available with the tests classifier. For example, this way Caffeine could reuse your QueueSanityTest rather than copying it in locally.