[Draft] High level design proposal for buffer pool in opensearch

[abiesps]

Is your feature request related to a problem? Please describe

In continuation to discussion on (#18833) and (#18838). I will continue to write a high level design proposal, followed by low level design and implementation choices on libraries/data structures and algorithms and another proposal for integration and tunability in opensearch.

Describe the solution you'd like

In this proposal, I'll focus on high level design of buffer pool in opensearch.

Functional requirements

1. Should support in-memory writes (writes to application pages) [ToDo]
2. Should support in-memory reads (reads from application pages)
3. Should support pluggable page replacement algorithms (evictions and prefetching).
4. Should support writing dirty pages to disk [ToDo]
5. Should support dynamic size allocations (different for reads/prefetch, readaheads and writes)
6. Should throttle reads/writes in case of memory pressure
7. Should support zero copy reads from application to storage device.
8. Must not result in data-loss (as long as application level sync/fsync is called) [ToDo]
9. Support delete after last read (Unix file system semantic of keeping data available for read till last reader is done even after delete, this behavior is seen during segment merges) [ToDo]

Non functional requirements

1. Low Read latency at high concurrency
2. Write throughput at high concurrency

Out of scope

1. Design for IO interface, will be covered separately.

Core Components

High level IO model with bufferpool

Memory Provider

Memory provider is going to be responsible for pre-allocating the memory outside of jvm heap (off-heap) memory. As part of low level details I’ll cover on possible memory provider implementations (for eg MemorySegments, sun.misc.Unsafe, netty etc ).

Page

Page is the unit of memory for all reads and writes. Page will be backed by off heap memory as returned from a memory region of bufferpool. Each page will have a constant size, which will depend on the operation for which the page is allocated for (write, read, prefetch, readahead etc). Each page is going to contain a page header and page content. Page header will contain page identifier (encoded with memory region index in the buffer pool), reference count, pin flag, page slice metadata information and a page level lock. Page level lock can be used to modify page header attributes (for instance reference count, pin flag etc). Page content are going to be used for writing or reading data. Once a page is flushed, page content will become immutable.

BufferPool

Buffer pool is going to provide an interface to allocate and deallocate/free pages. Internally BufferPool consists of multiple memory regions. Buffer pool is also going to be responsible for managing memory regions (resizing, fragmentation checks etc). I have to evaluate "range tree" versus hash table kind of core data structures for buffer pool [ToDo].

Memory Regions

Memory regions are the contiguous chunks of memory pre-allocated in the bufferpool. Each memory region will maintain its own set of data structures for tracking free pages, claimed pages and flushed pages. Since we are going to support multiple page sizes we may have to handle internal fragmentation in the buffer-pool. To simplify the handling of internal fragmentation, I am thinking of following options.

1. We can have dedicated memory region for each page size, this is simpler to implement but it can result in throttling, reads, writes even when we may have enough memory in different region. To make this simpler we can start with static size for each memory region (per size) and dynamically resize the regions based on usage in the background.
2. Other option is to have each memory region an affinity towards page size and buffer pool can try to allocate a page from preferred memory regions for the size. If there is no memory region for request size, then buffer-pool can try the memory region with next larger page size affinity. We will still have to handle the fragmentation.
   First option seems simpler to me, so possibly this is what I am going to go with for initial implementation.

BufferCache

BufferCache is going to store the mapping of filename and file block to the corresponding page in the bufferpool. Only pages that have successfully been fsynced are going to be part of buffer cache. Its possible that files that are not yet committed by index writer could be present in buffer cache because of periodic fsyncing of the pages in dirty page tracker. Index Reader will ensure not reading "non-committed" files, but buffer cache will not have these checks.

Dirty page tracker

Each memory region will have its own dirty page tracker that is going to keep track of all dirty pages and will be periodically and/or explicitly fsynced to the storage device.

Prefetching and readaheads

Prefetch requests are initiated by IndexInput. There will be background prefetcher threads that are going to be responsible for submitting prefetch requests. Prefetchers are also going to respect backpressure from memory(page replacement component) and io interface. All prefetch requests will be sent to IO interface for further processing.
Readaheads are going to be initiated from reads in IndexInput. Readaheads are also going to honor the memory and IO pressure as communicated by page replacement and io interface.
Both prefetch and readahead requests are going to be tied to the scope of an opensearch operation. (This will be discussed more in io interface design)

Page Replacement

Page replacement component is going to be responsible for managing the lifecycle of pages in the bufferpool as well as submitting quotas to prefetchers and readaheads to communicate memory pressure.

IOInterface

IOInterface is going to be a lower abstraction that does IO from a file identifier (file descriptor/channel) to the pages in the buffer pool. IO interface is also going to provide the capability of writing pages, reading pages, syncing pages to IO device.

User flows

Read flow

Write flow

Each memory region is going to maintain two list one for free pages and one for claimed pages. For pages that are going to be currently being written into are going to be owned by IndexOutput, till a flush is called on the page.
Flush is going to make the page content immutable and this page is going to be tracked in a different data structure, dirty page tracker as maintained by bufferpool.
fsync operation on a file is going to result in moving all pages associated with a file (and some more if io concurrency and batching allows for it) from dirty page tracker to buffer cache and storage device.
Buffer pool will also periodically fsync the pages tracked in dirty page tracker. This will make the page ready/available for reads.

Prefetch flow

1. All indexInput prefetch requests will be added to a queue of prefetcher.
2. Prefetchers are going to remove prefetch requests from the queue, if queue is not empty.
3. Prefetcher will have a prefetch quota from page replacement component and IO interface.
4. Each prefetcher thread is going to submit IO request based on the available quota and prefetch request scope validation from.

Readahead flow
Readahead flow is going to be similar to prefetch flow except there wont be an explicit call for readahead from index readers/index input. And readahead component will get a different quota for read ahead requests from page replacement component and io interface.

Periodic flushing/explicit sync from directory
There will be two distinct operations flush and fsync. Flush is going to result in moving the ownership of page allocated for write from IndexOutput to dirty page tracker. Flush will not guarantee that the page is written onto the storage device. Flush will be called at page granularity. Pages after flush may not be visible for read as they wont be added to buffer cache.
Buffer pool will also periodically submit write io for pages present in dirty page tracker and move them to buffer-cache.

There will be a file level fsync, which is going to flush all the dirty pages present for a file in dirty page tracker to storage device. Fsync request will only be sent to io interface once there is going to be an acknowledgement for all flush operations. Fsync operation will guarantee that all flushed pages for a file has been successfully written to io interface before calling fsync .

Related component

No response

Describe alternatives you've considered

No response

Additional context

No response

[jaredlundell]

My overall feedback on this design is that it's trying to solve too many problems all at once. We can build a much simpler system if we leverage the existing capabilities of the JVM instead of building things ourselves. If I were to sketch out how I'd build a buffer pool, it would be a Caffeine cache with a bounded size, using filepath+blockNumber as the cache keys and Direct ByteBuffers as the cache values. I don't think we need anything that's really more complicated than that.

Each page will have a constant size, which will depend on the operation for which the page is allocated for (write, read, prefetch, readahead etc)

Different sized pages for different use cases is going to make managing the buffer pool more difficult. It means we either need more book keeping and logic to keep track of what's in the cache, or we have effectively separate caches for the different use cases (which wouldn't be memory efficient). I'd suggest that we stick with a single, constant page size for the entire application. We can still do prefetches, etc at a higher layer in the application. For example, if we're using 4 KB pages and we want to do a 32 KB readahead, we can load 8 pages into the buffer pool.

Page header will contain page identifier (encoded with memory region index in the buffer pool), reference count, pin flag, page slice metadata information and a page level lock

Doing our own reference counting is going to be error prone. Is there a reason you don't want to just let the garbage collector handle this? And what's the use case for pinning? I think we can build a much simpler implementation if we don't try to manually manage everything about memory.

Internally BufferPool consists of multiple memory regions. Buffer pool is also going to be responsible for managing memory regions (resizing, fragmentation checks etc). I have to evaluate "range tree" versus hash table kind of core data structures for buffer pool [ToDo].

You're basically talking about writing our own memory allocator. I don't see a compelling reason to do that, as we're unlikely to do a significantly better job than the existing allocator. I would just do ByteBuffer.allocateDirect() and let the JVM deal with memory management. Anything more complicated than that seems like premature optimization.

Memory regions are the contiguous chunks of memory pre-allocated in the bufferpool. Each memory region will maintain its own set of data structures for tracking free pages, claimed pages and flushed pages. Since we are going to support multiple page sizes we may have to handle internal fragmentation in the buffer-pool.

This is a problem that we don't need to solve. Let the JVM deal with low-level memory allocation. It's pretty good at it.

Each memory region will have its own dirty page tracker that is going to keep track of all dirty pages and will be periodically and/or explicitly fsynced to the storage device.

This seems like complexity that doesn't need to be inside the buffer cache. Any reason why we can't leave this responsibility to the IndexWriter? Since we never actually overwrite the contents of any file, we should never actually have dirty pages.

[abiesps]

Thanks for the feedback @jaredlundell

Different sized pages for different use cases is going to make managing the buffer pool more difficult. It means we either need more book keeping and logic to keep track of what's in the cache, or we have effectively separate caches for the different use cases (which wouldn't be memory efficient). I'd suggest that we stick with a single, constant page size for the entire application. We can still do prefetches, etc at a higher layer in the application. For example, if we're using 4 KB pages and we want to do a 32 KB readahead, we can load 8 pages into the buffer pool.

I agree, as long as we are able to "batch" the reads at io layer as well. For instance 8 pages of 4kb each is consuming single IOPS. For most file systems kernel may already be doing this. In other cases, I am evaluating vectored read apis for the same.

Doing our own reference counting is going to be error prone. Is there a reason you don't want to just let the garbage collector handle this?

Opensearch has reference counted based caching of file block implementation in multiple places for e.g kraken plugin has it, writeable-warm (block level fetch) has it, so this is something we won't be starting from scratch. I agree that the block sizes are larger there (usually 8Mb) whereas here we are talking about pages but underlying logic is still the same. We can't rely on garbage collector to handle this as eviction from the cache will not be deterministic - we may end up removing a buffer that is currently being read.

And what's the use case for pinning ?

Need for pinning is going to be dependent on warm shards strategy, do we need to ensure any memory reservation for hot data. May not be needed in the initial cut.

You're basically talking about writing our own memory allocator. I don't see a compelling reason to do that, as we're unlikely to do a significantly better job than the existing allocator. I would just do ByteBuffer.allocateDirect() and let the JVM deal with memory management. Anything more complicated than that seems like premature optimization.

I don't have strong opinions on this, there are libraries that provide such allocators out of the box for eg. Netty has a pooled allocator for off-heap memory, I don’t think there are any such allocators for MemorySegments unless we always allocate memory from off heap for every new allocation. We have to measure the actual performance difference between preallocation and allocation per page to make this decision, if we choose to go with our own solution.

This seems like complexity that doesn't need to be inside the buffer cache. Any reason why we can't leave this responsibility to the IndexWriter? Since we never actually overwrite the contents of any file, we should never actually have dirty pages.

I think I shouldn't call them 'dirty pages' rather flushed pages (that are not yet fsynced). With buffered IO, IndexWriter maintains a buffer from heap in IndexOutput and keeps on writing to this buffer. Once the buffer is full it is flushed to BufferedOutputStream. With directIO Index writer does a file channel write once the buffer is full. Based on the performance difference between channel.write with direct IO and flush on bufferedOutputStream, we can take a call if we need this additional tracking or not, basically write throughput.

[andrross]

If I were to sketch out how I'd build a buffer pool, it would be a Caffeine cache with a bounded size, using filepath+blockNumber as the cache keys and Direct ByteBuffers as the cache values.

The idea of using an off-the-shelf cache we explored for the FileCache used by searchable snapshots for caching file blocks. There's some relevant discussion here. The key point is that these direct ByteBuffers are stateful and the cache can't evict an in-use ByteBuffer. It's probably worth taking another look at Caffeine to see whether the pinning feature can meet the needs though. Things may have changed in the last couple of years.

Somewhat similarly, I'm not sure we can rely on the garbage collector for cleanup. I think we need to be more deterministic with cleanup of the off-heap memory rather than waiting for GC to collect the on-heap reference.

And what's the use case for pinning ?

Need for pinning is going to be dependent on warm shards strategy.

I remain skeptical about the pinning functionality. Reference counting is pinning. If the thing has a reference count greater than zero, then it is pinned and not eligible for removal. The "pin flag" is just a special case of reference counting that doesn't count higher than 1. The pinning feature was recently added to the FileCache currently used by searchable snapshots. You can see the discussion here, but I remain unconvinced this was a good idea.

[jainankitk]

Thanks @abiesps for documenting the high level design. Few observations:

If I were to sketch out how I'd build a buffer pool, it would be a Caffeine cache with a bounded size, using filepath+blockNumber as the cache keys and Direct ByteBuffers as the cache values. I don't think we need anything that's really more complicated than that.

This is very similar to what we used for ultrawarm, minus the caffeine part. Although, given the much smaller page size (4kb) compared to ultrawarm (8MB), I am wondering if we need multi level cache similar to page table. I am assuming most of the non-heap memory will be available in the buffer pool, which can be easily few tens of GB, so ~10m entries in the cache

Each memory region will maintain its own set of data structures for tracking free pages, claimed pages and flushed pages. Since we are going to support multiple page sizes we may have to handle internal fragmentation in the buffer-pool.

I am wondering if we really need to solve the problem of having multiple page sizes initially. Optimally sharing and tracking memory across different page size pools will be additional overhead, without knowing exactly how beneficial it is going to be

Any reason why we can't leave this responsibility to the IndexWriter? Since we never actually overwrite the contents of any file, we should never actually have dirty pages.

This is a good point. While the index and directory themselves are read/write, individual segment files are never written to, so we should never have dirty pages

[jainankitk]

Reference counting is pinning. If the thing has a reference count greater than zero, then it is pinned and not eligible for removal. The "pin flag" is just a special case of reference counting that doesn't count higher than 1.

@andrross - Probably, I misunderstood something here. Isn't pinning to ensure that even the items with reference count never get evicted from the cache? Of course, we need to ensure that the pinned entries do not consume significant portion of cache

[andrross]

Reference counting is pinning. If the thing has a reference count greater than zero, then it is pinned and not eligible for removal. The "pin flag" is just a special case of reference counting that doesn't count higher than 1.

@andrross - Probably, I misunderstood something here. Isn't pinning to ensure that even the items with reference count never get evicted from the cache? Of course, we need to ensure that the pinned entries do not consume significant portion of cache

@jainankitk With pinning you can do something like:

refCountedThing.pin();
// do some things and can be sure that refCountedThing will never get destroyed
refCountedThing.unpin();
// now refCountedThing might get evicted and destroyed

That is functionally identical to:

refCountedThing.incRef();
// do some things and can be sure that refCountedThing will never get destroyed
refCountedThing.decRef();
// now refCountedThing might get evicted and destroyed

My point is with pinning you have to have some code that is responsible for pinning an item when you know you need to keep it, and then unpinning it when you no longer need to keep it. You could accomplish the same thing by using a reference count. You can read through the discussion here for the full details.

[jaredlundell]

The idea of using an off-the-shelf cache we explored for the FileCache used by searchable snapshots for caching file blocks. There's some #4964 (comment). The key point is that these direct ByteBuffers are stateful and the cache can't evict an in-use ByteBuffer.

@andrross The FileCache is a little bit different because it involves state on disk that needs explicit cleanup. A ByteBuffer is strictly an in-memory resource, which means the JVM's GC will automatically do the ref counting for us (effectively; obviously Java GCs don't actually do ref counting). If an application thread has obtained a reference to a ByteBuffer from the cache, the GC won't collect that ByteBuffer even if it's removed from the cache until the application thread is done with it. So if we're willing to let the GC manage all of our memory, we can have a really simple implementation. Given that the GC is responsible for managing all of our on-heap memory already, I'm not sure why we wouldn't trust it to also handle off-heap memory too. IMO the main reason to use off-heap memory for our buffer cache is to avoid the penalties that come with very large heap sizes (no Compressed OOPS, worse GC performance) rather than that we really need explicit control over memory lifetime.

[abiesps]

Thanks @jaredlundell , @andrross , @jainankitk for the feedback, initially I started the design with keeping constant page sizes in buffer-pool. I will try to explain the rationale of why I added that as one of the requirement.

I am wondering if we really need to solve the problem of having multiple page sizes initially. Optimally sharing and tracking memory across different page size pools will be additional overhead, without knowing exactly how beneficial it is going to be

I ran a fio benchmark (on 16xl instance to ensure network is not the bottleneck). My file system client has a maximum throughput limit of 1500MB/s and concurrency(slots) 64, with 4kb reads (iouring and direct io) I was only able to achieve a throughput of 44.2MB/s with io depth 64.

Therefore, we need a way to decouple io size from the 'block' size in the bufferpool. Since, we have use-cases for random access and sequential access both. We may not be able to drive the throughput we want with 4kb read size, especially for queries that need to access large data sequentially (large aggregations etc).

Initially I was exploring, if we can saturate the throughput by submitting X 4kb reads using ( https://man7.org/linux/man-pages/man3/io_uring_prep_readv.3.html ) but that also doesn't help as slots are allocated per read call. There is also overhead of each call. For instance in the file systems I am testing, the latency overhead of 32 4kb call versus single 128kb call is ~96ms and ~6ms.

So, there is a tradeoff here, imo we can decouple buffer pool with io size via keeping separate pool for submitting io requests based on the use-case (sequential reads, random reads, readaheads etc) and copy the data between bufferpool bytebuffer to io pools. But io pool will have to maintain pools for individual sizes.

Or we couple the buffer pool with io size and move the complexity to buffer-pool, since we anyway have to support dynamic sizing pool for io - this is also not a bad option.

[andrross]

Given that the GC is responsible for managing all of our on-heap memory already, I'm not sure why we wouldn't trust it to also handle off-heap memory too.

@jaredlundell The problems I see are related to the fact that we don't have control with the Java GC of when something gets collected after it goes out of scope. If we're accounting for native memory use by counting the objects we have in the on-heap cache, then if the ByteBuffer continues to live after removal from the cache then we can't rely on the cache size for enforcing limits on memory usage. We would risk exhausting system memory because the Java GC has no insight into how much off-heap memory our out-of-scope-but-not-yet-collected objects might be referencing.

Some of these problems could be mitigated by hooking into the java.lang.ref.Cleaner, but the problem remains that the cache's accounting would be inaccurate after an item is removed from the cache but still in use and referenced.