Skip to content

optimize LFU #438

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 8 commits into from
Nov 8, 2023
Merged

optimize LFU #438

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
fcdc1de
test
Oct 23, 2023
9914d9d
simplify+inline
Oct 23, 2023
491adec
rem del
Oct 23, 2023
9d8f937
Merge branch 'main' into users/alexpeck/lfuperf
bitfaster Oct 24, 2023
507e28a
inline iter
Oct 24, 2023
29a42cd
match caffeine
Nov 8, 2023
f638359
afterwrite
Nov 8, 2023
d1613fe
Merge branch 'main' into users/alexpeck/lfuperf
Nov 8, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 6 additions & 6 deletions BitFaster.Caching/Buffers/MpscBoundedBuffer.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,7 +86,7 @@ private int GetCount(int head, int tail)
public BufferStatus TryAdd(T item)
{
int head = Volatile.Read(ref headAndTail.Head);
int tail = Volatile.Read(ref headAndTail.Tail);
int tail = headAndTail.Tail;
int size = tail - head;

if (size >= buffer.Length)
Expand Down Expand Up @@ -117,7 +117,7 @@ public BufferStatus TryAdd(T item)
public BufferStatus TryTake(out T item)
{
int head = Volatile.Read(ref headAndTail.Head);
int tail = Volatile.Read(ref headAndTail.Tail);
int tail = headAndTail.Tail;
int size = tail - head;

if (size == 0)
Expand All @@ -136,7 +136,7 @@ public BufferStatus TryTake(out T item)
return BufferStatus.Contended;
}

Volatile.Write(ref buffer[index], null);
buffer[index] = null;
Volatile.Write(ref this.headAndTail.Head, ++head);
return BufferStatus.Success;
}
Expand Down Expand Up @@ -190,7 +190,7 @@ private int DrainToImpl(Span<T> output)
#endif
{
int head = Volatile.Read(ref headAndTail.Head);
int tail = Volatile.Read(ref headAndTail.Tail);
int tail = headAndTail.Tail;
int size = tail - head;

if (size == 0)
Expand All @@ -214,13 +214,13 @@ private int DrainToImpl(Span<T> output)
break;
}

Volatile.Write(ref localBuffer[index], null);
localBuffer[index] = null;
Write(output, outCount++, item);
head++;
}
while (head != tail && outCount < Length(output));

Volatile.Write(ref this.headAndTail.Head, head);
this.headAndTail.Head = head;

return outCount;
}
Expand Down
49 changes: 35 additions & 14 deletions BitFaster.Caching/Lfu/ConcurrentLfu.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
using System.Diagnostics;
using System.Diagnostics.CodeAnalysis;
using System.Linq;
using System.Runtime.CompilerServices;
using System.Runtime.InteropServices;
using System.Threading;
using System.Threading.Tasks;
Expand Down Expand Up @@ -70,7 +71,7 @@ public sealed class ConcurrentLfu<K, V> : ICache<K, V>, IAsyncCache<K, V>, IBoun

private readonly IScheduler scheduler;

private readonly LfuNode<K, V>[] drainBuffer;
private readonly LfuNode<K, V>[] drainBuffer;

/// <summary>
/// Initializes a new instance of the ConcurrentLfu class with the specified capacity.
Expand Down Expand Up @@ -478,9 +479,10 @@ private void AfterWrite(LfuNode<K, V> node)
private void ScheduleAfterWrite()
{
var spinner = new SpinWait();
int status = this.drainStatus.NonVolatileRead();
while (true)
{
switch (this.drainStatus.Status())
switch (status)
{
case DrainStatus.Idle:
this.drainStatus.Cas(DrainStatus.Idle, DrainStatus.Required);
Expand All @@ -494,6 +496,7 @@ private void ScheduleAfterWrite()
{
return;
}
status = this.drainStatus.VolatileRead();
break;
case DrainStatus.ProcessingToRequired:
return;
Expand All @@ -509,8 +512,8 @@ IEnumerator IEnumerable.GetEnumerator()

private void TryScheduleDrain()
{
if (this.drainStatus.Status() >= DrainStatus.ProcessingToIdle)
{
if (this.drainStatus.NonVolatileRead() >= DrainStatus.ProcessingToIdle)
{
return;
}

Expand All @@ -521,15 +524,15 @@ private void TryScheduleDrain()

if (lockTaken)
{
int status = this.drainStatus.Status();
int status = this.drainStatus.NonVolatileRead();

if (status >= DrainStatus.ProcessingToIdle)
{
return;
}

this.drainStatus.Set(DrainStatus.ProcessingToIdle);
scheduler.Run(() => DrainBuffers());
this.drainStatus.VolatileWrite(DrainStatus.ProcessingToIdle);
scheduler.Run(() => this.DrainBuffers());
}
}
finally
Expand Down Expand Up @@ -559,15 +562,15 @@ private void DrainBuffers()
}
}

if (this.drainStatus.Status() == DrainStatus.Required)
if (this.drainStatus.VolatileRead() == DrainStatus.Required)
{
TryScheduleDrain();
}
}

private bool Maintenance(LfuNode<K, V> droppedWrite = null)
{
this.drainStatus.Set(DrainStatus.ProcessingToIdle);
this.drainStatus.VolatileWrite(DrainStatus.ProcessingToIdle);

// Note: this is only Span on .NET Core 3.1+, else this is no-op and it is still an array
var buffer = this.drainBuffer.AsSpanOrArray();
Expand Down Expand Up @@ -609,10 +612,10 @@ private bool Maintenance(LfuNode<K, V> droppedWrite = null)
// 1. We drained both input buffers (all work done)
// 2. or scheduler is foreground (since don't run continuously on the foreground)
if ((done || !scheduler.IsBackground) &&
(this.drainStatus.Status() != DrainStatus.ProcessingToIdle ||
(this.drainStatus.NonVolatileRead() != DrainStatus.ProcessingToIdle ||
!this.drainStatus.Cas(DrainStatus.ProcessingToIdle, DrainStatus.Idle)))
{
this.drainStatus.Set(DrainStatus.Required);
this.drainStatus.NonVolatileWrite(DrainStatus.Required);
}

return done;
Expand Down Expand Up @@ -743,13 +746,15 @@ private ref struct EvictIterator
public LfuNode<K, V> node;
public int freq;

[MethodImpl(MethodImplOptions.AggressiveInlining)]
public EvictIterator(CmSketch<K> sketch, LfuNode<K, V> node)
{
this.sketch = sketch;
this.node = node;
freq = node == null ? -1 : sketch.EstimateFrequency(node.Key);
}

[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void Next()
{
node = node.Next;
Expand Down Expand Up @@ -863,9 +868,10 @@ private class DrainStatus

private PaddedInt drainStatus; // mutable struct, don't mark readonly

[MethodImpl(MethodImplOptions.AggressiveInlining)]
public bool ShouldDrain(bool delayable)
{
int status = Volatile.Read(ref this.drainStatus.Value);
int status = this.NonVolatileRead();
return status switch
{
Idle => !delayable,
Expand All @@ -875,19 +881,34 @@ public bool ShouldDrain(bool delayable)
};
}

public void Set(int newStatus)
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void VolatileWrite(int newStatus)
{
Volatile.Write(ref this.drainStatus.Value, newStatus);
}

[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void NonVolatileWrite(int newStatus)
{
this.drainStatus.Value = newStatus;
}

[MethodImpl(MethodImplOptions.AggressiveInlining)]
public bool Cas(int oldStatus, int newStatus)
{
return Interlocked.CompareExchange(ref this.drainStatus.Value, newStatus, oldStatus) == oldStatus;
}

public int Status()
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public int VolatileRead()
{
return Volatile.Read(ref this.drainStatus.Value);
}

[MethodImpl(MethodImplOptions.AggressiveInlining)]
public int NonVolatileRead()
{
return this.drainStatus.Value;
}

[ExcludeFromCodeCoverage]
Expand Down