fix(buffers): Batch delete

[ktff]

Ref. #7425

Reduces performance impact of deletes. In worst case scenario when sink manages to stay up to date, every event would result in one delete operation which negatively impacted write/read performance. This PR batches deletes to reduce that.

[ktff]

This isn't fully related too batching but it is a performance win which is kinda related.

[blt]

I'm confused about why we're doing mem::take here. What's going on is we're allocating a fresh, zero-length VecDeque, extending the taken buffer by the contents of the value_iter and then placing the taken buffer back into self.buffer, trashing the freshly allocated VecDeque. Is this to work around a lifetime issue with sending the buffer down into block_in_place?

Could we not move the call to extend to outside the block_in_place?

[ktff]

Is this to work around a lifetime issue with sending the buffer down into block_in_place?

Exactly. The compiler is having issue with mutable and immutable borrow at the same time from self in that function, while creating fresh VecDeque is just copying bytes around, there is even a possibility that the compiler is optimizing it away.

Could we not move the call to extend to outside the block_in_place?

By having extend in block_in_place we also cover the possibility that the iterator is lazy and actually accessing disk while iterating. I haven't checked for this behavior but either way it seems as a good insurance.

[blt]

Gotcha. At this point it seems like there are a couple of suppositions I'd like to get empirical data for:

• the compiler is able to optimized the temporary VecDeque (I find this believable)
• the iterator is lazy and doing extend internal to block_in_place is a performance boon

This function is going to be called enough that I'd like to be sure. We have benchmarks in place for disk buffering specifically, so measuring and approach with and without the temporary would be sufficient to my mind.

[ktff]

A, it seams VecDeque::new is allocating... yea I'll see what can be changed. At worst I'll extract this change to a separate PR for benchmarking.

[blt]

Cool, here's benchmarks for the disk buffer on my system. I had to rebase with master but otherwise the code here is unchanged.

[blt]

Here's the benchmark with the following diff applied:

diff --git a/lib/vector-core/buffers/src/disk/leveldb_buffer/reader.rs b/lib/vector-core/buffers/src/disk/leveldb_buffer/reader.rs
index 840cef35c..0f24033e8 100644
--- a/lib/vector-core/buffers/src/disk/leveldb_buffer/reader.rs
+++ b/lib/vector-core/buffers/src/disk/leveldb_buffer/reader.rs
@@ -106,16 +106,14 @@ where
             // This will usually complete instantly, but in the case of a large
             // queue (or a fresh launch of the app), this will have to go to
             // disk.
-            let mut buffer = std::mem::take(&mut self.buffer);
-            tokio::task::block_in_place(|| {
-                buffer.extend(
-                    self.db
-                        .value_iter(ReadOptions::new())
-                        .from(&Key(self.read_offset))
-                        .to(&Key(self.read_offset + 100)),
-                );
+            let vals: VecDeque<Vec<u8>> = tokio::task::block_in_place(|| {
+                self.db
+                    .value_iter(ReadOptions::new())
+                    .from(&Key(self.read_offset))
+                    .to(&Key(self.read_offset + 100))
+                    .collect()
             });
-            self.buffer = buffer;
+            self.buffer.extend(vals);
         }

         if let Some(value) = self.buffer.pop_front() {

[blt]

From these measurements it looks like the median runtimes don't shift all that much for both write-then-read and write-and-read but the distribution is more favorable to the diff version I posted, especially in write-and-read where we'd be hitting the empty condition frequently. You can even see that the modified version I made did fewer overall iterations per benchmark because the results were more stable, which tracks. Removing an allocation from a hot path will give that path more consistent runtime, generally speaking.

[ktff]

The compiler is having issue with mutable and immutable borrow at the same time from self

I've digged for a bit and found it's caused by Pin. That's fixed now so this works as expected, without intermediates.

[blt]

Nice work here.