Skip to content

Commit

Permalink
optimize LFU (#438)
Browse files Browse the repository at this point in the history 
* test

* simplify+inline

* rem del

* inline iter

* match caffeine

* afterwrite

---------
  • Loading branch information
@bitfaster
bitfaster authored Nov 8, 2023
1 parent d3a781c commit a99aa78
Show file tree
Hide file tree
Showing 2 changed files with 41 additions and 20 deletions.
12 changes: 6 additions & 6 deletions BitFaster.Caching/Buffers/MpscBoundedBuffer.cs
View file
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
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

0 comments on commit a99aa78

Please sign in to comment.