Eliminate array copies #31

wolfchimneyrock · 2023-09-11T21:38:45Z

*Issue number of the reported bug or feature request: #30

Describe your changes

BrokerSession passes Collection instead of PutMessageImpl[] to avoid making copies.
ApplicationData calculates crc32c on the fly to avoid having to make an expensive copy of the whole payload.
ByteBufferOutputStream:
- add peek() call to make an independent view of the underlying data.
- don't copy and repack ByteBuffer's written to the ByteBufferOutputStream.
NettyTcpConnection still makes a copy of data read, but let netty make the copy instead of us.

Testing performed
Unit and integration tests have been updated to reflect the changes.

Additional context
As discussed with @sgalichkin

…copies

sgalichkin

Decided to provide some feedback while still looking at ByteBufferOutputStream and its usage

sgalichkin · 2023-09-12T11:05:37Z

bmq-sdk/src/main/java/com/bloomberg/bmq/impl/QueueImpl.java

-        brokerSession.post(this, msgs);
-        return msgs;
+    public void flush() throws BMQException {
+        brokerSession.post(this, putMessages.getAndSet(new ArrayList<>(INITIAL_PUTMESSAGES_SIZE)));


Should not we check here if there are messages to send? Otherwise we may just replace empty array with another empty array

sgalichkin · 2023-09-12T11:13:05Z

bmq-sdk/src/main/java/com/bloomberg/bmq/impl/QueueImpl.java

-        synchronized (lock) {
-            putMessages.add(message);
-        }
+        putMessages.get().add(message);


I suppose .get().add() is not an atomic operation, so it's possible get the array which is being flushed and add new messages afterward and eventually lost them or modify the collection while it is being iterated in PutPoster?

.get() itself is atomic, and so is the .getAndSet() in flush, so afaik I don't think its possible to get the array that is being flushed

Agree, get() and getAndSet() are atomic, but the following sequence seems to be possible:
[t1] .get() returns an array to flush
[t2] .getAndSet() returns the same array
[t2] is iterating the array and processing the messages
[t1] calls .add() which modifies the array

Correct?

sgalichkin · 2023-09-12T11:16:04Z

bmq-sdk/src/main/java/com/bloomberg/bmq/impl/BrokerSession.java

@@ -1249,6 +1253,14 @@ public void post(QueueHandle queueHandle, PutMessageImpl... msgs) throws BMQExce
        }
    }

+    public void post(QueueHandle queueHandle, PutMessageImpl msg) throws BMQException {


Do we need there there two methods here?

Collections.singletonList() is more efficient for a single element than Arrays.asList()

sgalichkin · 2023-09-12T11:16:34Z

bmq-sdk/src/main/java/com/bloomberg/bmq/impl/PutPoster.java

        pack(msgs);
        flush();
    }

+    public void post(PutMessageImpl msg) {


Same question here

sgalichkin · 2023-09-12T11:34:23Z

bmq-sdk/src/main/java/com/bloomberg/bmq/impl/infr/io/ByteBufferOutputStream.java

    public void writeBytes(String s) throws IOException {
        throw new UnsupportedOperationException();
    }

-    public void writeBytes(ByteBuffer b) throws IOException {
+    private boolean bufferIsFresh(ByteBuffer b) {


Maybe rename to bufferIsClear()?

sgalichkin · 2023-09-12T13:29:18Z

bmq-sdk/src/main/java/com/bloomberg/bmq/impl/infr/io/ByteBufferOutputStream.java

+                .toArray(ByteBuffer[]::new);
+    }
+
+    private ArrayList<ByteBuffer> buffers() {


Is this method used?

yeah maybe I can get rid of it

sgalichkin · 2023-09-12T13:33:45Z

bmq-sdk/src/main/java/com/bloomberg/bmq/impl/infr/io/ByteBufferOutputStream.java

+     */
+    public ByteBuffer[] peek() {
+        return bbArray.stream()
+                .map(bb -> (ByteBuffer) (bb.duplicate().flip()))


Maybe asReadOnlyBuffer() instead of duplicate()?

I tried that first, but a lot of unit tests then fail with:

[ERROR] com.bloomberg.bmq.impl.infr.proto.SchemaEventImplBuilderTest.testBuildOpenQueueJson Time elapsed: 0.001 s <<< ERROR! java.nio.ReadOnlyBufferException at java.base/java.nio.ByteBuffer.array(ByteBuffer.java:1473) at com.bloomberg.bmq.util.TestHelpers.buffersContents(TestHelpers.java:65) at com.bloomberg.bmq.util.TestHelpers.compareWithFileContent(TestHelpers.java:91) at com.bloomberg.bmq.impl.infr.proto.SchemaEventImplBuilderTest.testBuildOpenQueueJson(SchemaEventImplBuilderTest.java:152) at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77) at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.base/java.lang.reflect.Method.invoke(Method.java:568) at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:59) at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12) at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:56) at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17) at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306) at org.junit.runners.BlockJUnit4ClassRunner$1.evaluate(BlockJUnit4ClassRunner.java:100) at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:366) at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:103) at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:63) at org.junit.runners.ParentRunner$4.run(ParentRunner.java:331) at org.apache.maven.surefire.junitcore.pc.Scheduler$1.run(Scheduler.java:345) at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) at java.base/java.lang.Thread.run(Thread.java:833) ```

It looks like the peek() method is called by gson. Is it possible to mark the method to be ignored when serializing?

sgalichkin · 2023-09-12T14:30:40Z

bmq-sdk/src/main/java/com/bloomberg/bmq/impl/infr/io/ByteBufferOutputStream.java

    public void writeBytes(String s) throws IOException {
        throw new UnsupportedOperationException();
    }

-    public void writeBytes(ByteBuffer b) throws IOException {
+    private boolean bufferIsFresh(ByteBuffer b) {
+        // a buffer that has never been put() to nor flipped


Not completely sure if this statement is correct. From javadocs:
clear() makes a buffer ready for a new sequence of channel-read or relative put operations: It sets the limit to the capacity and the position to zero.

So it is possible to put some data (up to limit which may be set to capacity value) and then call clear() in order to be able to read it.

Or my understanding is incorrect and this is a requirement for ByteBuffer argument to be "fresh"?

this is one thing that really bugs me about nio.ByteBuffer, there is no way to determine if a buffer that has no data vs one that has data but has already been flipped for reading. isClear() is probably a better name for this method ...

but in the long term, I think the better thing to do is replace every usage of nio.ByteBuffer with netty.ByteBuf in the entire SDK, which would

allow to eliminate that initial copy from NettyTcpConnection

allow to delete ByteBufferOutputStream and ByteBufferInputStream entirely since netty.ByteBufOutputStream can be used instead

this is one thing that really bugs me about nio.ByteBuffer, there is no way to determine if a buffer that has no data vs one that has data but has already been flipped for reading. isClear() is probably a better name for this method ...

Agree, but maybe we do not need to determine this and just take what is between 0 and limit regardless of current position?

but in the long term, I think the better thing to do is replace every usage of nio.ByteBuffer with netty.ByteBuf in the entire SDK, which would

allow to eliminate that initial copy from NettyTcpConnection

allow to delete ByteBufferOutputStream and ByteBufferInputStream entirely since netty.ByteBufOutputStream can be used instead

I would agree but AFAIK the limited use of netty and its components was intentional. The reason is that in some day SDK may switch to another network library and in case of wide use of its components in the code e.g. netty.ByteBuf instead of nio.ByteBuffer etc. it would require a lot of work to do. And having netty.ByteBuf in API class (Queue) will make it even more difficult.

I suggest to have a quick call tomorrow to discuss this

sgalichkin · 2023-09-12T14:33:18Z

bmq-sdk/src/main/java/com/bloomberg/bmq/impl/infr/io/ByteBufferOutputStream.java

-                continue;
-            }
-            getNewBuffer();
+        if (bufferIsFresh(b)) return;


So taking into account the comment above, is this statement correct?

sgalichkin · 2023-09-18T13:07:44Z

bmq-sdk/src/main/java/com/bloomberg/bmq/impl/infr/net/NettyTcpConnection.java

+                // TODO - can this copy be eliminated?
+                ByteBuffer nioBuf = ByteBuffer.allocate(byteBuf.readableBytes());
+                byteBuf.readBytes(nioBuf);
+                readBuffer.writeBuffer(nioBuf);


So here we make a temporary buffer which gets bytes from netty one and then is added to the output stream as is without any copies?

Maybe it makes sense to add a comment saying this.
And maybe I would preserve a part of the removed comment in order to explain why we need to make a copy in general

wolfchimneyrock · 2023-09-22T15:37:09Z

I was able to run the JMH benchmarks with this latest commit, here are the results

Before:

Benchmark                                               Mode  Cnt     Score     Error  Units
ApplicationDataBenchmark.testZlibStreamInOut           thrpt   10     8.664 ±   0.379  ops/s
SessionBenchmark.sendReceive512B                       thrpt    5  1690.186 ±  76.546  ops/s
SessionBenchmark.sendReceive512B_Zlib                  thrpt    5  1686.011 ± 115.191  ops/s
SessionBenchmark.sendReceive512KiB                     thrpt    5     6.356 ±   0.110  ops/s
SessionBenchmark.sendReceive512KiB_Zlib                thrpt    5     6.360 ±   0.184  ops/s
SessionBenchmark.sendReceive5MiB                       thrpt    5    12.180 ±   0.367  ops/s
SessionBenchmark.sendReceive5MiB_Zlib                  thrpt    5    15.087 ±   0.116  ops/s
SessionBenchmark.sendReceive60MiB                      thrpt    5     1.013 ±   0.037  ops/s
SessionBenchmark.sendReceive60MiB_Zlib                 thrpt    5     1.358 ±   0.021  ops/s
SessionBenchmark.sendReceiveBatch1000ZlibConfirmLater  thrpt    5     0.169 ±   0.003  ops/s
SessionBenchmark.sendReceiveBatch1000ZlibConfirmNow    thrpt    5     0.225 ±   0.014  ops/s
SessionBenchmark.sendReceiveBatch100ConfirmLater       thrpt    5     1.539 ±   0.093  ops/s
SessionBenchmark.sendReceiveBatch100ConfirmNow         thrpt    5     2.155 ±   0.056  ops/s
SessionBenchmark.sendReceiveBatch100ZlibConfirmNow     thrpt    5     2.120 ±   0.088  ops/s
SessionBenchmark.sendReceiveBatch800ZlibConfirmLater   thrpt    5     0.197 ±   0.003  ops/s

After:

Benchmark                                               Mode  Cnt     Score     Error  Units
ApplicationDataBenchmark.testZlibStreamInOut           thrpt   10     6.088 ±   0.569  ops/s
SessionBenchmark.sendReceive512B                       thrpt    5  1881.041 ± 178.831  ops/s
SessionBenchmark.sendReceive512B_Zlib                  thrpt    5  1868.423 ± 186.251  ops/s
SessionBenchmark.sendReceive512KiB                     thrpt    5    18.013 ±   0.627  ops/s
SessionBenchmark.sendReceive512KiB_Zlib                thrpt    5    16.877 ±   0.652  ops/s
SessionBenchmark.sendReceive5MiB                       thrpt    5    34.717 ±   0.714  ops/s
SessionBenchmark.sendReceive5MiB_Zlib                  thrpt    5    16.922 ±   0.130  ops/s
SessionBenchmark.sendReceive60MiB                      thrpt    5     3.089 ±   0.121  ops/s
SessionBenchmark.sendReceive60MiB_Zlib                 thrpt    5     1.503 ±   0.031  ops/s
SessionBenchmark.sendReceiveBatch1000ZlibConfirmLater  thrpt    5     0.181 ±   0.003  ops/s
SessionBenchmark.sendReceiveBatch1000ZlibConfirmNow    thrpt    5     0.238 ±   0.009  ops/s
SessionBenchmark.sendReceiveBatch100ConfirmLater       thrpt    5     8.246 ±   0.714  ops/s
SessionBenchmark.sendReceiveBatch100ConfirmNow         thrpt    5     8.279 ±   0.657  ops/s
SessionBenchmark.sendReceiveBatch100ZlibConfirmNow     thrpt    5     2.242 ±   0.074  ops/s
SessionBenchmark.sendReceiveBatch800ZlibConfirmLater   thrpt    5     0.226 ±   0.005  ops/s

except for the individual 512B messages, looks like 2x - 4x higher throughput for uncompressed.
for ZLib, I think there is still some bottleneck as the increase isn't so pronounced.

wolfchimneyrock added 3 commits September 8, 2023 11:11

pass Collection<PutMessageImpl> instead of PutMessageImpl[] to avoid …

18fffad

…copies

pack() doesn't need synchronization with AtomicReference

cfd4ec0

remove extra copies with ByteBufferOutputStream, crc calculations

ae2cd67

sgalichkin self-assigned this Sep 12, 2023

wolfchimneyrock force-pushed the eliminate-array-copies branch from f2917cc to 1467cbe Compare September 12, 2023 20:17

more efficient buffer usage, add tests

f20e57d

wolfchimneyrock force-pushed the eliminate-array-copies branch from 1467cbe to f20e57d Compare September 12, 2023 20:22

sgalichkin reviewed Sep 18, 2023

View reviewed changes

wolfchimneyrock added 2 commits September 18, 2023 11:24

tweak buffer slicing

471bfe3

ApplicationData payload is bbos, direct duplicate from bbis to bbos

0b157ac

review comments

a7de790

sgalichkin assigned 678098 and unassigned sgalichkin Mar 15, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Eliminate array copies #31

Eliminate array copies #31

wolfchimneyrock commented Sep 11, 2023

sgalichkin left a comment

sgalichkin Sep 12, 2023

sgalichkin Sep 12, 2023

wolfchimneyrock Sep 19, 2023

sgalichkin Sep 19, 2023

sgalichkin Sep 12, 2023

wolfchimneyrock Sep 22, 2023

sgalichkin Sep 12, 2023

sgalichkin Sep 12, 2023

sgalichkin Sep 12, 2023

sgalichkin Sep 12, 2023

wolfchimneyrock Sep 19, 2023

sgalichkin Sep 12, 2023

wolfchimneyrock Sep 19, 2023

sgalichkin Sep 19, 2023

sgalichkin Sep 12, 2023

wolfchimneyrock Sep 19, 2023

sgalichkin Sep 19, 2023 •

edited

Loading

sgalichkin Sep 12, 2023

sgalichkin Sep 18, 2023 •

edited

Loading

wolfchimneyrock commented Sep 22, 2023 •

edited

Loading

Eliminate array copies #31

Are you sure you want to change the base?

Eliminate array copies #31

Conversation

wolfchimneyrock commented Sep 11, 2023

sgalichkin left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sgalichkin Sep 19, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sgalichkin Sep 18, 2023 • edited Loading

Choose a reason for hiding this comment

wolfchimneyrock commented Sep 22, 2023 • edited Loading

sgalichkin Sep 19, 2023 •

edited

Loading

sgalichkin Sep 18, 2023 •

edited

Loading

wolfchimneyrock commented Sep 22, 2023 •

edited

Loading