Removes old buffer pool #16

buritos · 2020-01-09T23:55:05Z

Removes ByteBufferPool;
Fixes bugs related to the release of ConsumerByteBuffer more than once. PoolAwareConsumerByteBuffer::release is made idempotent and logs warning when it is called on the same instance more than once;
Changes the release pattern for releasing ConsumerByteBuffer when it is handed over to an async operation. The buffer is released on caught exceptions but not in a finally block, since that would most likely run in parallel with the async operation.

Possible issues with merging this PR for upstream:

ConsumerByteBuffer leaks from the pool due to not being properly released in async operations implemented upstream. This is not too big of a deal since the new pool is unbounded and therefore leaked buffers will not cause an availability issue. However, leaking buffers interferes with the pool's ability to scale down and failure to release buffers will eventually cause memory issues (large footprint, more frequent GC). See inline comments for places where we should investigate.
Signatures of constructors and methods where the old pool was used have been changed to accept the new pool type.
Direct upstream dependencies to the old pool.

…fix)

…onsumer::consume

src/main/java/io/vlingo/wire/channel/RequestChannelConsumer.java

src/main/java/io/vlingo/wire/channel/ResponseChannelConsumer.java

src/main/java/io/vlingo/wire/fdx/bidirectional/rsocket/RSocketClientChannel.java

buritos · 2020-01-09T23:58:33Z

src/main/java/io/vlingo/wire/fdx/bidirectional/rsocket/RSocketClientChannel.java

      this.logger = logger;
    }

    private void handle(Payload payload) {
-      final ByteBufferPool.PooledByteBuffer pooledBuffer = readBufferPool.accessFor("client-response", 25);
+      final ConsumerByteBuffer pooledBuffer = readBufferPool.acquire();
      try {
        final ByteBuffer payloadData = payload.getData();
        final ConsumerByteBuffer put = pooledBuffer.put(payloadData);


Can this be avoided?

Ditto above.

buritos · 2020-01-09T23:59:48Z

src/main/java/io/vlingo/wire/fdx/bidirectional/rsocket/RSocketChannelContext.java

@@ -81,14 +83,13 @@ public void close() {
  }

  public void consume(Payload request) {
-    final ByteBufferPool.PooledByteBuffer pooledBuffer = readBufferPool.accessFor("client-request");
+    final ConsumerByteBuffer pooledBuffer = readBufferPool.acquire();
    try {
      pooledBuffer.put(request.getData());


Can this be avoided?

Ditto above.

buritos · 2020-01-10T00:00:56Z

src/main/java/io/vlingo/wire/fdx/bidirectional/SecureClientRequestResponseChannel.java

@@ -209,7 +202,7 @@ public void onSuccess() {

    @Override
    public void onInput(final ByteBuffer decrypted) {
-      final ConsumerByteBuffer buffer = readBufferPool.accessFor("ssl-read");
+      final ConsumerByteBuffer buffer = readBufferPool.acquire();
      consumer.consume(buffer.put(decrypted).flip());


Can put be avoided?

Do you mean that the copy is unnecessary? Otherwise, what is the problem you see and how would you do it otherwise? BTW, I did not implement the RSocket support so I am a bit out of context here. Thanks for explaining a bit more.

I am concerned about the allocation overhead of copying potentially large buffers. There is a documented case here: https://github.com/buritos/vlingo-wire/blob/master/src/main/java/io/vlingo/wire/fdx/bidirectional/rsocket/RSocketClientChannel.java#L73, but I am not sure about the implications of not making a copy and also if it is the same case in all of these methods I mentioned in this PR. What I don't understand is how copying the buffer addresses "payload might not be sent immediately". I can of-course investigate, but it would be faster is someone could explain the intent. Also, how can we be sure that the original buffer fits in the buffer that we get from the pool? The way I would remove the copying is by adding a constructor in ConsumerByteBuffer creates an instance from an existing ByteBuffer, instead of allocating a new one. Let me know if you remember anything that would help me understand this code faster than reading it and what's your thoughts about my assumptions so far.

@buritos I like your idea of reusing a buffer, but I don't know if the copy-to buffer is owned by RSocket or vlingo. @alexguzun implemented the RSocket components so he can explain why this is done that way. He is in Lisbon and you can ping him on our team Slack workspace.

Hi! The copy-to buffer ar owned by wire. They are required because unfortunatelly the rsocket and wire threads are separate meaning that a wire outbound might try to send the second message while the RSocket is still sending the first one. If we use the same buffer, the second message will overrite the first one.

Thank you for explaining @alexguzun. I will spend some more time on this now that I clearly understand your motivation for copying. We might be able to get around this by pushing the acquisition of the buffers all the way up, reusing the buffers from the pool instead of holding on to particular instances. Does that make sense? Any thoughts?

VaughnVernon

@buritos In general this looks like what we need. I will await your clarification on the put(...) issues as you see them.

VaughnVernon · 2020-01-10T01:52:19Z

src/main/java/io/vlingo/wire/message/ConsumerByteBufferPool.java

+    void activate() {
+      active.set(true);
+    }
+
    @Override
    public String toString() {
      return "PooledByteBuffer[id=" + id() + "]";
    }

    @Override
    public void release() {


@buritos Nice and simple.

src/main/java/io/vlingo/wire/fdx/bidirectional/rsocket/RSocketClientChannel.java

VaughnVernon · 2020-01-10T02:01:18Z

src/main/java/io/vlingo/wire/fdx/bidirectional/SecureClientRequestResponseChannel.java

@@ -209,7 +202,7 @@ public void onSuccess() {

    @Override
    public void onInput(final ByteBuffer decrypted) {
-      final ConsumerByteBuffer buffer = readBufferPool.accessFor("ssl-read");
+      final ConsumerByteBuffer buffer = readBufferPool.acquire();
      consumer.consume(buffer.put(decrypted).flip());


Do you mean that the copy is unnecessary? Otherwise, what is the problem you see and how would you do it otherwise? BTW, I did not implement the RSocket support so I am a bit out of context here. Thanks for explaining a bit more.

VaughnVernon · 2020-01-10T02:01:33Z

src/main/java/io/vlingo/wire/fdx/bidirectional/rsocket/RSocketChannelContext.java

@@ -81,14 +83,13 @@ public void close() {
  }

  public void consume(Payload request) {
-    final ByteBufferPool.PooledByteBuffer pooledBuffer = readBufferPool.accessFor("client-request");
+    final ConsumerByteBuffer pooledBuffer = readBufferPool.acquire();
    try {
      pooledBuffer.put(request.getData());


Ditto above.

VaughnVernon · 2020-01-10T02:01:48Z

src/main/java/io/vlingo/wire/fdx/bidirectional/rsocket/RSocketClientChannel.java

      this.logger = logger;
    }

    private void handle(Payload payload) {
-      final ByteBufferPool.PooledByteBuffer pooledBuffer = readBufferPool.accessFor("client-response", 25);
+      final ConsumerByteBuffer pooledBuffer = readBufferPool.acquire();
      try {
        final ByteBuffer payloadData = payload.getData();
        final ConsumerByteBuffer put = pooledBuffer.put(payloadData);


Ditto above.

buritos · 2020-01-10T10:25:59Z

Turns out that we are hitting the bug that was discovered: https://travis-ci.org/vlingo/vlingo-wire/builds/635019329#L2492

So happy we found this!

Thanks for assigning the task @VaughnVernon ❤️

This reverts commit 62570e8.

buritos · 2020-01-10T14:21:16Z

@buritos In general this looks like what we need. I will await your clarification on the put(...) issues as you see them.

@VaughnVernon, my theory is that all the copying can be eliminated with some rearrangement of the code that performs the release. My gut feeling is that the code is structured in this way (eager to release) because of the limitations of the previous pool implementation (failure to meet demand). Given the scalable nature of the new implementation (we pay the price for it), it is ok to hold on to a buffer instance for longer, meaning that it is ok to release it at the final consumer, as long as the producers do not attempt to reuse it in any way (they should always acquire another buffer from the pool).

@alexguzun confirms (although a bit concerned about the rippling effects of changing the API for ManagedOutboundChannel::write). I believe we can work around that, preserving semantics of write(ByteBuffer) by overloading write to also accept ConusmerByteBuffer, then implement write(ByteBuffer) in terms of write(ConsumerByteBuffer), all of which are very local to wire.

Let me know if you think that the performance gains from such change are worth it to prioritize the changes now, or something that you would like to see in the future, or not at all :)

VaughnVernon · 2020-01-10T14:42:45Z

With the reverted test enablement is the bug is not fix, or you are working on this further?

VaughnVernon · 2020-01-10T14:53:54Z

Also, we definitely need to fix this. The copying is *expensive * and in this wire API is used in multiple components, or could be. I think the current changes and bug fixes you have made and that we benefit from immediately should be merged today. The tuning should be entered as an issue and fixed soon, but not immediately.

buritos · 2020-01-10T14:59:55Z

With the reverted test enablement is the bug is not fix, or you are working on this further?

I would like to try the tests again after eliminating catch(Throwable) from all the places.

buritos · 2020-01-10T15:09:17Z

Also, we definitely need to fix this. The copying is *expensive * and in this wire API is used in multiple components, or could be. I think the current changes and bug fixes you have made and that we benefit from immediately should be merged today. The tuning should be entered as an issue and fixed soon, but not immediately.

I've made an issue. You are welcome to merge this in its current state. I haven't checked upstream for buffer release in all the places I removed it from here. However, it is low risk due to the nature of the new pool (allocates as necessary). The only issue is that leaks interfere with scaling down the pool after bursts.

VaughnVernon · 2020-01-10T15:57:26Z

The only issue is that leaks interfere with scaling down the pool after bursts.

Merge done. Could you elaborate why/how this happens?

buritos · 2020-01-10T16:05:37Z

The InUse counter is not decremented, tricking the pool to think that there is more demand than there actually is (leaked buffers are garbage collected).

VaughnVernon · 2020-01-10T19:44:47Z

Got it, thanks.

buritos added 8 commits January 9, 2020 14:49

prevents releasing pooled byte buffer more than once per lease (quick…

3fd2b2c

…fix)

documents convention for releasing buffers passed to ResponseChannelC…

8342cc0

…onsumer::consume

removes old buffer pool from BasicClientRequestResponseChannel

aff910e

removes old buffer pool from RSocketChannelContext

6702f86

removes old buffer pool from RSocketClientChannel

984d29a

removes old buffer pool from SecureClientRequestResponseChannel

609501e

removes old buffer pool from ApplicationOutboundStream

4d185ad

deletes old byte buffer pool

45711f3

buritos commented Jan 9, 2020

View reviewed changes

src/main/java/io/vlingo/wire/channel/RequestChannelConsumer.java Show resolved Hide resolved

buritos commented Jan 9, 2020

View reviewed changes

src/main/java/io/vlingo/wire/channel/ResponseChannelConsumer.java Show resolved Hide resolved

buritos commented Jan 9, 2020

View reviewed changes

src/main/java/io/vlingo/wire/fdx/bidirectional/rsocket/RSocketClientChannel.java Show resolved Hide resolved

buritos commented Jan 9, 2020

View reviewed changes

buritos commented Jan 10, 2020

View reviewed changes

VaughnVernon reviewed Jan 10, 2020

View reviewed changes

buritos added 7 commits January 10, 2020 11:29

enables RSocket tests in travis

62570e8

Revert "enables RSocket tests in travis"

5b67ab2

This reverts commit 62570e8.

includes some of the RSocket tests

aa6bae8

triggers build

89c855f

ignores only the test that consistently fails

be593a5

excludes all RSocketClientChannelTest tests

a96f590

excludes all RSocket tests

33df030

buritos mentioned this pull request Jan 10, 2020

Too much overhead from copying ByteBuffers #17

Closed

VaughnVernon merged commit 8eb034f into vlingo:master Jan 10, 2020

buritos deleted the remove-old-buffer-pool branch January 13, 2020 08:56

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Removes old buffer pool #16

Removes old buffer pool #16

buritos commented Jan 9, 2020

buritos Jan 9, 2020 •

edited

VaughnVernon Jan 10, 2020

buritos Jan 9, 2020

VaughnVernon Jan 10, 2020

buritos Jan 10, 2020

VaughnVernon Jan 10, 2020

buritos Jan 10, 2020

VaughnVernon Jan 10, 2020

alexguzun Jan 10, 2020

buritos Jan 10, 2020

VaughnVernon left a comment

VaughnVernon Jan 10, 2020

VaughnVernon Jan 10, 2020

VaughnVernon Jan 10, 2020

VaughnVernon Jan 10, 2020

buritos commented Jan 10, 2020 •

edited

buritos commented Jan 10, 2020

VaughnVernon commented Jan 10, 2020

VaughnVernon commented Jan 10, 2020

buritos commented Jan 10, 2020

buritos commented Jan 10, 2020

VaughnVernon commented Jan 10, 2020

buritos commented Jan 10, 2020 via email

VaughnVernon commented Jan 10, 2020

Removes old buffer pool #16

Removes old buffer pool #16

Conversation

buritos commented Jan 9, 2020

buritos Jan 9, 2020 • edited

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

VaughnVernon 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

buritos commented Jan 10, 2020 • edited

buritos commented Jan 10, 2020

VaughnVernon commented Jan 10, 2020

VaughnVernon commented Jan 10, 2020

buritos commented Jan 10, 2020

buritos commented Jan 10, 2020

VaughnVernon commented Jan 10, 2020

buritos commented Jan 10, 2020 via email

VaughnVernon commented Jan 10, 2020

buritos Jan 9, 2020 •

edited

buritos commented Jan 10, 2020 •

edited