Improve performance of reading large result sets #835
Conversation
Codecov Report
@@ Coverage Diff @@
## main #835 +/- ##
==========================================
+ Coverage 84.83% 85.01% +0.17%
==========================================
Files 126 126
Lines 1728 1742 +14
Branches 202 147 -55
==========================================
+ Hits 1466 1481 +15
+ Misses 262 261 -1
... and 1 file with indirect coverage changes 📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more |
| val withTimeout: F[Chunk[Byte]] => F[Chunk[Byte]] = readTimeout match { | ||
| private val withTimeout: F[Option[Chunk[Byte]]] => F[Option[Chunk[Byte]]] = readTimeout match { | ||
| case _: Duration.Infinite => identity | ||
| case finite: FiniteDuration => _.timeout(finite) | ||
| } |
There was a problem hiding this comment.
.timeout(...) now supports Duration.Infinite so this helper shouldn't be needed anymore.
There was a problem hiding this comment.
Yeah I tried that but it benchmarked a bit slower since the short circuit happens once with current implementation but on each call with the new CE support.
There was a problem hiding this comment.
I'm amazed that makes such a difference 😅
|
|
||
| private def readUntilN(nBytes: Int, carry: Chunk[Byte]): F[BitVector] = | ||
| if (carry.size < nBytes) { | ||
| withTimeout(socket.read(8192)).flatMap { |
There was a problem hiding this comment.
Should this parameter be customizeable?
There was a problem hiding this comment.
I played with increasing it and it seemed to make no difference fwiw.
There was a problem hiding this comment.
Yeah I tried a bunch of values and settled on same value fs2 uses in reads.
There was a problem hiding this comment.
Might also need to play with SO_RCVBUF option to see an effect.
There was a problem hiding this comment.
I hope it's ok that I add my comments just as a visitor/user.
I noticed a performance improvement only when setting SND/RCV buffers to a really huge value (8mb). This assumes that the session pool is reasonably small (8-64, for instance.). And it's probably a good idea to have it configurable.
There was a problem hiding this comment.
Comments very welcome! The socket options are configurable now. The read size here isn’t but we could make it so. I’d like to show a case where it matters though before breaking compatibility by introducing a new param. Same for the background queue size in BufferedMessageSocket, which is hardcoded to 256 and doesn’t seem to fill.
There was a problem hiding this comment.
Also, I suspect receive buffer size may be more influential on performance when connecting to a remote postgres database as opposed to local host.
| bvs.read(5).flatMap { bits => | ||
| val (tag, len) = header.decodeValue(bits).require | ||
| val tag = bits.take(8).toByte() |
There was a problem hiding this comment.
Are you just inlining the header decoder? Or why this change?
There was a problem hiding this comment.
Avoids allocations of a bunch of intermediate objects. Makes a small bit consistent improvement on the benchmark.
| // This software is licensed under the MIT License (MIT). | ||
| // For more information see LICENSE or https://opensource.org/licenses/MIT | ||
|
|
||
| package test |
There was a problem hiding this comment.
I think this should be tests right? A
There was a problem hiding this comment.
It’s copied from QueryTest but yeah not sure why we have both test and tests.
There was a problem hiding this comment.
Maybe this is too daring: BitVector is used to decode messages in the code above (and of course elsewhere). Maybe it would improve performance using ByteVector. To my knowledge the PG wire protocol's smallest unit is one byte. And ByteVector is really highly optimized. It would require a huge refactoring, though. The refactored code would be a lot simpler dealing with bytes instead of bits.
There was a problem hiding this comment.
There shouldn’t be a significant difference and scodec’s Codec abstraction is built on BitVector.
Fixes #833