From 56f80380f059740990239d2f008640ba7bb853c9 Mon Sep 17 00:00:00 2001 From: Peter Veentjer Date: Wed, 15 Jan 2025 09:48:55 +0200 Subject: [PATCH] AtomicBuffer opaque operations Added opaque operations to AtomicBuffer. An opaque operation provides atomicity and visibility, but it doesn't provide any ordering guarantees beyond coherence. So it doesn't order loads/stores to different addresses, only to its own address. Opaque operations are great for performance counters or progress indicators and provide the least amount of overhead on the CPU (all modern CPUs are coherent); especially CPUs with a weak memory model like ARM or RISC-V. --- .../org/agrona/concurrent/AtomicBuffer.java | 56 ++++++++++++ .../org/agrona/concurrent/UnsafeBuffer.java | 85 +++++++++++++++++++ 2 files changed, 141 insertions(+) diff --git a/agrona/src/main/java/org/agrona/concurrent/AtomicBuffer.java b/agrona/src/main/java/org/agrona/concurrent/AtomicBuffer.java index 989446c1..4dcbb928 100644 --- a/agrona/src/main/java/org/agrona/concurrent/AtomicBuffer.java +++ b/agrona/src/main/java/org/agrona/concurrent/AtomicBuffer.java @@ -148,6 +148,34 @@ public interface AtomicBuffer extends MutableDirectBuffer */ long addLongOrdered(int index, long increment); + /** + * Atomically put a value to a given index with opaque semantics. + * + * @param index in bytes for where to put. + * @param value for at a given index. + */ + void putLongOpaque(int index, long value); + + /** + * Atomically get a value to a given index with opaque semantics. + * + * @param index in bytes for where to put. + * @return the value for at a given index. + */ + long getLongOpaque(int index); + + /** + * Adds a value to a given index with opaque semantics. The read and write + * will be atomic, but the combination is not atomic. So don't use this + * method concurrently because you can run into lost updates due to + * a race-condition. + * + * @param index in bytes for where to put. + * @param increment by which the value at the index will be adjusted. + * @return the previous value at the index. + */ + long addLongOpaque(int index, long increment); + /** * Atomically adds a value to a given index with ordered store semantics. Use a negative increment to decrement. *

@@ -270,6 +298,34 @@ public interface AtomicBuffer extends MutableDirectBuffer */ int addIntRelease(int index, int increment); + /** + * Atomically put a value to a given index with opaque semantics. + * + * @param index in bytes for where to put. + * @param value for at a given index. + */ + void putIntOpaque(int index, int value); + + /** + * Atomically get a value to a given index with opaque semantics. + * + * @param index in bytes for where to put. + * @return the value for at a given index. + */ + int getIntOpaque(int index); + + /** + * Adds a value to a given index with opaque semantics. The read and write + * will be atomic, but the combination is not atomic. So don't use this + * method concurrently because you can run into lost updates due to + * a race-condition. + * + * @param index in bytes for where to put. + * @param increment by which the value at the index will be adjusted. + * @return the previous value at the index. + */ + int addIntOpaque(int index, int increment); + /** * Atomic compare and set of an int given an expected value. *

diff --git a/agrona/src/main/java/org/agrona/concurrent/UnsafeBuffer.java b/agrona/src/main/java/org/agrona/concurrent/UnsafeBuffer.java index 355c61d4..f46e76c0 100644 --- a/agrona/src/main/java/org/agrona/concurrent/UnsafeBuffer.java +++ b/agrona/src/main/java/org/agrona/concurrent/UnsafeBuffer.java @@ -453,6 +453,48 @@ public long addLongRelease(final int index, final long increment) return UnsafeApi.getAndAddLongRelease(byteArray, addressOffset + index, increment); } + /** + * {@inheritDoc} + */ + public void putLongOpaque(final int index, final long value) + { + if (SHOULD_BOUNDS_CHECK) + { + boundsCheck0(index, SIZE_OF_LONG); + } + + UnsafeApi.putLongOpaque(byteArray, addressOffset + index, value); + } + + /** + * {@inheritDoc} + */ + public long getLongOpaque(final int index) + { + if (SHOULD_BOUNDS_CHECK) + { + boundsCheck0(index, SIZE_OF_LONG); + } + + return UnsafeApi.getLongOpaque(byteArray, addressOffset + index); + } + + /** + * {@inheritDoc} + */ + public long addLongOpaque(final int index, final long increment) + { + if (SHOULD_BOUNDS_CHECK) + { + boundsCheck0(index, SIZE_OF_LONG); + } + + final long oldValue = UnsafeApi.getLongOpaque(byteArray, addressOffset + index); + final long newValue = oldValue + increment; + UnsafeApi.putLongOpaque(byteArray, addressOffset + index, newValue); + return oldValue; + } + /** * {@inheritDoc} */ @@ -573,6 +615,49 @@ public int addIntRelease(final int index, final int increment) return UnsafeApi.getAndAddIntRelease(byteArray, addressOffset + index, increment); } + + /** + * {@inheritDoc} + */ + public void putIntOpaque(final int index, final int value) + { + if (SHOULD_BOUNDS_CHECK) + { + boundsCheck0(index, SIZE_OF_LONG); + } + + UnsafeApi.putIntOpaque(byteArray, addressOffset + index, value); + } + + /** + * {@inheritDoc} + */ + public int getIntOpaque(final int index) + { + if (SHOULD_BOUNDS_CHECK) + { + boundsCheck0(index, SIZE_OF_LONG); + } + + return UnsafeApi.getIntOpaque(byteArray, addressOffset + index); + } + + /** + * {@inheritDoc} + */ + public int addIntOpaque(final int index, final int increment) + { + if (SHOULD_BOUNDS_CHECK) + { + boundsCheck0(index, SIZE_OF_LONG); + } + + final int oldValue = UnsafeApi.getIntOpaque(byteArray, addressOffset + index); + final int newValue = oldValue + increment; + UnsafeApi.putIntOpaque(byteArray, addressOffset + index, newValue); + return oldValue; + } + /** * {@inheritDoc} */