Exposing FASTER log as a first-class abstraction

[badrishc]

We are exposing our high-speed latch-free log (or write-ahead log) facility as a generally usable first-class citizen, called FasterLog. You can append variable-sized chunks (as byte[] or Span<byte>) to the log. There is no additional header in the log. Users can perform (potentially tailing) pull-based iteration over any range of the log (including until a future log address). Users can truncate from the head of the log. They can also flush the log at very fine granularity, lower than the usual per-page flushing. All aspects of the log (page size, number of pages in memory, segment size on disk, etc.) are tunable. The log works with async code as well, and we will add async versions of the API going forward.

Note that there is no random lookup index on top of this log - if you want such an indexed log, use FasterKv as usual.

Improvements to Epochs

As part of this PR, we have also significantly optimized the epoch framework, removing important performance bottlenecks that now make it feasible to acquire and release epochs at a very fine granularity. We can perform 35 million acquire/release operation pairs per second per thread, with linear scalability. If users are able to pre-acquire a thread for the operations, we are able to achieve 100 million acquire/release operation pairs per second per thread. This improvement now makes it possible to make FASTER work at high performance in task-based async environments.

[badrishc]

See example at https://github.com/microsoft/FASTER/blob/19d5d82641965df3b203b2b9e141ba0594051977/cs/playground/FasterLogSample/Program.cs

[badrishc]

Note that this log can live on top of any device that implements the IDevice abstraction. We provide out of the box device implementations for:

• Generic local storage device (any .NET core system)
• Local storage device (Windows-specific high performance unbuffered overlapped IO)
• Azure page blob storage device
• Tiered storage device (create tiers of storage, log writes to all tiers, read from lowest available tier)
• Sharded storage device (spread writes across multiple devices to get higher effective bandwidth)

[badrishc]

Checked in log commit and recovery support. By default, we flush the log at page boundaries. But, the user can manually flush more frequently, e.g., every 10ms. The playground sample has been updated to show this facility as well. You can kill and restart the sample as many times as you want, it will auto-resume.

[badrishc]

As for end to end performance, for 100 byte payloads, we get around 1GB/sec to a modern NVMe SSD with a single thread, with log tail increment being the bottleneck as we increase the number of append threads. For 1000 byte payloads, we get close to the SSD bandwidth limit (1.8 - 2 GB/sec) using one thread.

[hiteshmadan]

instead of pinning the entry, you can use the CopyTo API (which is also apparently faster in many cases than other memory copy APIs).

entry.CopyTo(new Span((void*)(4 + physicalAddress), length));

[badrishc]

I actually benchmarked the two alternatives and went with pinning as it was non-trivially faster.

[hiteshmadan]

In Iterator.GetNext, there is a code path that allocates a byte[] and copies the read data to it.

1. Does this happen when the read and write are happening on the same shared buffer? If so, it'd be valuable to allow for the allocation to be via a memory pool since this is going to be a pretty frequent occurrence (reader keeping up with the writer).
   Please consider accepting an IMemoryPool and rent an array from that pool. You'll probably have to return an IMemoryOwner instead of a Span though.

2. In code paths where it doesn't allocate, is the span safe to keep around as a reference in the app code? Or does it point to a buffer that (eventually) can be reused for a different page?

[tli2]

Anticipating the meta-device package, this static factory will not be sufficient to encapsulate the variety of device created. Certainly not every device will care about all of these flags.

Maybe instead of adding another boolean flag, we should start moving away from it?

[badrishc]

That makes sense, but the goal was to unblock current users because the recovery path was causing slowdown and incorrectness in certain cases. The goal was to make "no recovery" as the default so current users default to that.

[tli2]

Do we need to truncate the segments that are not tracked by this device?

[badrishc]

You mean delete existing files when the device is created? That would depend upon whether the device is being used as new or for recovery.

[tli2]

Maybe I am missing something, but it's unclear to me why the list is required. Is it just for readability?

[badrishc]

The list of files comes in alphabetical order by default, not numerical. Caused a bug without sorting.

[tli2]

I am wondering if byte[] is too low level of an abstraction. Perhaps an interface with a custom serialization/deserialization method is a better choice?

[badrishc]

I thought of that and even tried to prototype it but it was not as convenient in the end because the user had to implement too many callbacks. This can be revisited though.

[hiteshmadan]

+1 to having a batch append API. However this API forces too many allocations - 1 for the list, and 1 array allocation for each entry in this case:
If the serialized byte array is generated by a serializer that pools its' buffers (like Microsoft Bond), most likely it'll return the result as an ArraySegment / Span / IMemoryOwner.

How does something like this look:

(note that Span cannot be used here in place of ArraySegment, details here: https://adamsitnik.com/Span/#span-must-not-be-a-generic-type-argument)

(this can be further optimized to ask the caller for the totalLength as a function parameter and do only one BlockAllocate outside the for loop, not sure if that's worth it)

public unsafe bool TryAppend<TState>(int numEntries, Func<int, TState, ArraySegment<byte>> getBytes, TState state, out long logicalAddress)
        {
            epoch.Resume();
            logicalAddress = 0;
            long tail = -allocator.GetTailAddress();
            allocator.CheckForAllocateComplete(ref tail);
            if (tail < 0)
            {
                epoch.Suspend();
                return false;
            }

            for (int i = 0; i < numEntries; i++)
            {
                Span<byte> entry = getBytes(i, state);
                var length = entry.Length;
                BlockAllocate(4 + length, out logicalAddress);
                var physicalAddress = allocator.GetPhysicalAddress(logicalAddress);
                *(int*)physicalAddress = length;
                entry.CopyTo(new Span<byte>((void*)(4 + physicalAddress), length));
            }

            epoch.Suspend();
            return true;
        }

[hiteshmadan]

As an aside - Does the batch partially show up to the iterator while its still being written? Or does the reader only see the first event from the batch after epoch.Suspend() ?

[AlgorithmsAreCool]

@hiteshmadan
I don't understand everything thats going on in this PR, but if you want to use a span instead of an array segment you just need to declare a custom delegate like

public delegate ReadOnlySpan<byte> ByteGetter<TState>(int index, TState state);

or something like it. That sidesteps the generic parameter constraint.

[badrishc]

After all the optimizations and inlining, there isn't much benefit to a batched interface in terms of performance, because acquire and release epoch - which were the only calls getting amortized - have been made very inexpensive. Also, we can't allocate for the batch in bulk because the page sizes are pretty small. Maybe we should stick to the unbatched interface?

[hiteshmadan]

If there is already pending data waiting to be flushed + the incoming entry is big enough to not fit on the space remaining on the current page + no more pages can be allocated, BlockAllocate will block the thread waiting for the allocation to go through.

Is there any way for the early-exit check to account for entry.Length too?

[badrishc]

We only made the common path non blocking right now. Do we need it to always be non blocking? The system atomically allocates space when when it's not yet ready to use, then waits for it to become usable. If we want it to be non blocking, we will need to surface this in the API, i.e. TryAppend will return an incomplete address that the user will need to compete by calling TryCompleteAppend. Is this what you are looking for?

[badrishc]

In Iterator.GetNext, there is a code path that allocates a byte[] and copies the read data to it.

1. Does this happen when the read and write are happening on the same shared buffer? If so, it'd be valuable to allow for the allocation to be via a memory pool since this is going to be a pretty frequent occurrence (reader keeping up with the writer).
   Please consider accepting an IMemoryPool and rent an array from that pool. You'll probably have to return an IMemoryOwner instead of a Span though.
2. In code paths where it doesn't allocate, is the span safe to keep around as a reference in the app code? Or does it point to a buffer that (eventually) can be reused for a different page?

Correct, when reads and writes share the buffer, we have to copy out because we can only access memory safely under epoch protection. Agreed that memory pool is a possibility. If we had that, a cleaner design might be to always copy out to the user-provided buffer regardless of where we are reading from. This will address your concern in point 2 as well (currently, you must give up the span as soon as you invoke another GetNext).

[badrishc]

Fyi, there is a check-in coming in to reduce the number of pages needed in memory. This required a couple of internal changes for correctness.

[badrishc]

We now have both low-mem support (as low as 4 pages) as well as truly non-blocking TryAppend.

We also have an async API variant for append (AppendAsync) that returns only when append flush is done, but this is a rough API that needs further feedback/review. There are also significant performance implications of using this interface. Find the prototype here: #180

[badrishc]

This PR is almost ready to merge, and definitely ready to try out. Features include:

(1) Support for blocking Append, TryAppend, and AppendAsync
(2) Support for appending byte[] and Span to log
(3) Log auto-commit at page boundary + user-controlled commit (e.g., every 5ms)
(4) Configurable commit provider (ILogCommitManager) that can write commit info anywhere, and hook in pre- and post-commit operations.
(5) Atomic commit of a batch of entries to log (ISpanBatch)
(6) Multi-threaded and multi-task appends supported
(7) Tailing iterator over the log
(8) Exposed CommitTask so users can await the next commit if they want
(9) Support for truncation of log (from the head)

See https://github.com/microsoft/FASTER/blob/master/docs/cs/FasterLog.md for a first draft of usage guide, and the comprehensive sample with comments at https://github.com/microsoft/FASTER/blob/fasterlog/cs/playground/FasterLogSample/Program.cs

Pending

We need to refine/enhance the ways that pooled Span can be provided to the iterator; currently we defined a delegate called GetMemory here. Any other ideas here?

[badrishc]

Summary of Interface to FasterLog

// Enqueue log entry (to memory) with spain-wait

long Enqueue(byte[] entry)
long Enqueue(ReadOnlySpan<byte> entry)
long Enqueue(IReadOnlySpanBatch readOnlySpanBatch)

// Try to enqueue log entry (to memory)

bool TryEnqueue(byte[] entry, out long logicalAddress)
bool TryEnqueue(ReadOnlySpan<byte> entry, out long logicalAddress)
bool TryEnqueue(IReadOnlySpanBatch readOnlySpanBatch, out long logicalAddress)

// Async enqueue log entry (to memory)

async ValueTask<long> EnqueueAsync(byte[] entry)
async ValueTask<long> EnqueueAsync(ReadOnlyMemory<byte> entry)
async ValueTask<long> EnqueueAsync(IReadOnlySpanBatch readOnlySpanBatch)

// Wait for commit

void WaitForCommit(long untilAddress = 0) // spin-wait
async ValueTask WaitForCommitAsync(long untilAddress = 0)

// Commit

void Commit(bool spinWait = false)
async ValueTask CommitAsync()

// Helper: enqueue log entry and spin-wait for commit

long EnqueueAndWaitForCommit(byte[] entry)
long EnqueueAndWaitForCommit(ReadOnlySpan<byte> entry)
long EnqueueAndWaitForCommit(IReadOnlySpanBatch readOnlySpanBatch)

// Helper: enqueue log entry and async wait for commit

async ValueTask<long> EnqueueAndWaitForCommitAsync(byte[] entry)
async ValueTask<long> EnqueueAndWaitForCommitAsync(ReadOnlyMemory<byte> entry)
async ValueTask<long> EnqueueAndWaitForCommitAsync(IReadOnlySpanBatch readOnlySpanBatch)

// Truncate log (from head)

void TruncateUntil(long untilAddress)

// Scan interface

FasterLogScanIterator Scan(long beginAddress, long endAddress)

// FasterLogScanIterator interface

bool GetNext(out byte[] entry, out int entryLength)
bool GetNext(MemoryPool<byte> pool, out IMemoryOwner<byte> entry, out int entryLength)
async ValueTask WaitAsync()

// Random read

async ValueTask<(byte[], int)> ReadAsync(long address, int estimatedLength = 0)

[badrishc]

Added checksum support for log verification during scan/read. Enable by setting FasterLogSettings.LogChecksum.

[badrishc]

Update: we are in the final stages of this PR. Working on correctly handling and surfacing exceptional cases, such as transient and permanent storage failures. This is WIP in a branch (fasterlog-exceptions). ETA within a couple of days.

Edit: exception support is taking a bit longer due to subtle corner cases involving parallel flush and error conditions. Will update here when merged.

[badrishc]

IAsyncEnumerable support has now been added for the iterator.

[badrishc]

In addition to the previously supported TruncateUntil on the log, we now support persistent iterators. You can create any number of iterators over the log, and "name" them if you need them to be part of commits. During recovery, if you create an iterator with the same name, we will resume iteration from the last committed iterator location. Example:

using (iter = log.Scan(log.BeginAddress, long.MaxValue, name: "foo"))
   await foreach ((byte[] result, int length) in iter.GetAsyncEnumerable())
   {
      ...
   }

[badrishc]

Detailed documentation is now available at:

• https://github.com/microsoft/FASTER/blob/master/docs/cs/FasterLog.md
• https://microsoft.github.io/FASTER/docs/fasterlog

[badrishc]

Merging to master and closing PR as the functionality is complete at this point. We can continue discussions here, and create new PRs for further enhancements as well.