Improve throughput of read streams by transferring multiple records at once

[vweevers]

Working on the level-bench benchmarks got me thinking. Currently level-iterator-stream ignores the size argument of stream._read(size). Per tick it transfers only 1 db record from the underlying iterator to the stream's buffer. I think we can be smarter about this. By connecting the knowledge that size records are requested, all the way down to the db (in the case of leveldown, down to the C++, potentially replacing its current read-ahead cache mechanism).

In pseudo-code it would look something like (ignore error handling for a moment):

// level-iterator-stream
ReadStream.prototype._read = function (size) {
  var self = this

  // Fetch <size> records from the db, then call "visit" repeatedly within a tick
  this._iterator.visit(size, function visit (record) {
     // Record is either null, an object { key, value } or just a key or value
     self.push(record)
  })
})

This also avoids allocating 3 callback functions per record. Alternatively:

this._iterator.nextv(size, function (records) { // aka "chunks" in streams
   for (let record of records) self.push(record)
})

Or if streams were to get a .pushv method similar to .writev:

this._iterator.nextv(size, function (records) {
   self.pushv(records)
})

/cc @mcollina: could such an API be faster? I'm also wondering how _read() behaves in an asyncIterator. Is size always 1 in that case, or does the stream read ahead?

@peakji @ralphtheninja /cc @kesla

[peakji]

Respecting size is good for streams, but I guess moving from eager read-ahead cache to on demand read-ahead cache won't make much difference, because it cannot reduce the number of times we cross the C++ / JS boundary?

[mcollina]

Getting more than one item at the same time will significantly increase throughput.

Is size always 1 in that case, or does the stream read ahead?

The streams read ahead accordingly to highWaterMark. Essentially it would try to read 16 entries by default.

---

On-demand read head will guarantee some performance speedup because it reduces the time an object will live on the heap: as a result, the GC will end up doing less work. I would recommend implementing this anyway.

[vweevers]

Error-handling included, I propose the following API:

iterator.nextv(size, function (err, records) {
   if (err) // errored
   if (records.length === 0) // reached end

   for (let record of records) // ..
})

• If size < 0 then nextv yields all records (or as much as limit) (there is no default safeguard)
• If size === 0 then nextv yields 0 records (it's illegal to call nextv(0) more than once)
• nextv respects the boolean keys and values options passed to the iterator ctor. If both are true, record will be a { key, value } object. Else record will be either a key or value.

[vweevers]

We can play with this idea in leveldown: introduce a nextv method on the iterator, combine it with a temporary fork of level-iterator-stream that utilizes nextv, then benchmark it.

It might increase the number of times we cross the C++ / JS boundary, especially for small records, because the highWaterMark of streams in objectMode is measured in number of records, while the highWaterMark of leveldown's iterator is measured in bytes. IMO this does not matter because both these parameters can be tweaked by the user as necessary. Would it warrant semver-major though?

[MeirionHughes]

what changes to leveldown would this require? changing the c++ iterator cache size (per iterator) to match the nextv(size... ?

[vweevers]

Depends on what we want to do with the highWaterMark (in bytes) logic. I'm now wondering, in current code, whether db.createReadStream({ highWaterMark }) passes highWaterMark to both the iterator and the stream. That would be a problem.

[Raynos]

👍

We implement iterator.batchNext() instead of iterator.nextv() in a userland library. We use the undocumented iterator.cache field.

When creating a leveldown iterator we set the high water mark to 1024 * 1024 aka 1MB to make sure the cache is populated.

Implementing a native nextv() in leveldown so that we don't access the private .cache field would be nice.

[MeirionHughes]

I didn't even know you could specify it on the iterator options (its not on the docs). Is highWaterMark meant to be not documented? Also is this something that is (or can be) made uniform across AbstractLevelDown implementations?