Skip to content

delete buffer #233

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

Draft
wants to merge 9 commits into
base: main
Choose a base branch
from
Draft

delete buffer #233

Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
5da6149
multirem
Sep 4, 2022
0269a53
pad
Sep 4, 2022
7074a7d
buff
Sep 5, 2022
26e4e98
merge
Sep 8, 2022
7121e47
buffer
Sep 8, 2022
3da2e1f
deletebuffer
Sep 8, 2022
f256098
Merge branch 'main' into users/alexpeck/multirem
Sep 9, 2022
bdbf913
merge
Sep 10, 2022
247c16e
Merge branch 'main' into users/alexpeck/multirem
Sep 17, 2022
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
172 changes: 170 additions & 2 deletions BitFaster.Caching/Lfu/ConcurrentLfu.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,9 @@ public sealed class ConcurrentLfu<K, V> : ICache<K, V>, IAsyncCache<K, V>, IBoun

private readonly IScheduler scheduler;

private readonly MpmcBoundedBuffer<List<LfuNode<K, V>>> cleanBufferQueue = new MpmcBoundedBuffer<List<LfuNode<K, V>>>(Environment.ProcessorCount);
private readonly MpmcBoundedBuffer<List<LfuNode<K, V>>> deleteBufferQueue = new MpmcBoundedBuffer<List<LfuNode<K, V>>>(Environment.ProcessorCount);

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

/// <summary>
Expand Down Expand Up @@ -181,6 +184,7 @@ public void Trim(int itemCount)
{
// flush all buffers
Maintenance();
DeleteAllBufferedItems();

// walk in lru order, get itemCount keys to evict
TakeCandidatesInLruOrder(this.probationLru, candidates, itemCount);
Expand Down Expand Up @@ -334,8 +338,49 @@ private void AfterWrite(LfuNode<K, V> node)
}

TryScheduleDrain();
TryDeleteBufferedItems();
}

//////////////////////////////////////////////////////////////////////////////////////////////////

//while (true)
//{
// bool wasTaken = false;
// Monitor.TryEnter(this.maintenanceLock, ref wasTaken);
// try
// {
// if (wasTaken)
// {
// // aggressively try to exit the lock early before doing full maintenance
// var status = BufferStatus.Contended;
// while (status != BufferStatus.Full)
// {
// status = writeBuffer.TryAdd(node);

// if (status == BufferStatus.Success)
// {
// ScheduleAfterWrite();
// return;
// }
// }

// Maintenance(node);
// break;
// }
// }
// finally
// {
// if (wasTaken)
// {
// Monitor.Exit(this.maintenanceLock);
// }
// }
//}

//DeleteAllBufferedItems();

//////////////////////////////////////////////////////////////////////////////////////////////////

lock (this.maintenanceLock)
{
// aggressively try to exit the lock early before doing full maintenance
Expand All @@ -354,6 +399,9 @@ private void AfterWrite(LfuNode<K, V> node)
// if the write was dropped from the buffer, explicitly pass it to maintenance
Maintenance(node);
}

// remove outside the lock
DeleteAllBufferedItems();
}

private void ScheduleAfterWrite()
Expand Down Expand Up @@ -433,6 +481,9 @@ private void DrainBuffers()
done = Maintenance();
}

// remove outside the lock
DeleteAllBufferedItems();

// don't run continuous foreground maintenance
if (!scheduler.IsBackground)
{
Expand All @@ -450,6 +501,7 @@ private bool Maintenance(LfuNode<K, V> droppedWrite = null)
{
this.drainStatus.Set(DrainStatus.ProcessingToIdle);


// extract to a buffer before doing book keeping work, ~2x faster
int readCount = readBuffer.DrainTo(this.drainBuffer);

Expand Down Expand Up @@ -480,6 +532,8 @@ private bool Maintenance(LfuNode<K, V> droppedWrite = null)
}

EvictEntries();
QueueDeleteBuffer();

this.capacity.OptimizePartitioning(this.metrics, this.cmSketch.ResetSampleSize);
ReFitProtected();

Expand All @@ -496,6 +550,119 @@ private bool Maintenance(LfuNode<K, V> droppedWrite = null)
return done;
}

private void TryAddToRemoveList(LfuNode<K, V> node)
{
deleteBuffer.Add(node);

if (deleteBuffer.Count >= deleteBufferSize)
{
// int attempts = 0;

while (true)
{
if (this.deleteBufferQueue.TryAdd(deleteBuffer) == BufferStatus.Success)
{
if (this.cleanBufferQueue.TryTake(out deleteBuffer) == BufferStatus.Success)
{
return;
}

deleteBuffer = new List<LfuNode<K, V>>(deleteBufferSize);
return;
}
}

//foreach (var n in deleteBuffer)
//{
// this.dictionary.TryRemove(n.Key, out var _);
// Disposer<V>.Dispose(n.Value);
//}

//deleteBuffer.Clear();
}
}

private void QueueDeleteBuffer()
{
if (deleteBuffer.Count == 0)
{
return;
}

while (true)
{
if (this.deleteBufferQueue.TryAdd(deleteBuffer) == BufferStatus.Success)
{
if (this.cleanBufferQueue.TryTake(out deleteBuffer) == BufferStatus.Success)
{
return;
}

deleteBuffer = new List<LfuNode<K, V>>(deleteBufferSize);
return;
}
}

//if (this.deleteBufferQueue.TryAdd(deleteBuffer) != BufferStatus.Success)
//{
// foreach (var n in deleteBuffer)
// {
// this.dictionary.TryRemove(n.Key, out var _);
// Disposer<V>.Dispose(n.Value);
// }

// deleteBuffer.Clear();
//}
//else
//{
// if (this.cleanBufferQueue.TryTake(out deleteBuffer) == BufferStatus.Success)
// {
// return;
// }

// deleteBuffer = new List<LfuNode<K, V>>(deleteBufferSize);
//}
}

const int deleteBufferSize = 128;
private List<LfuNode<K, V>> deleteBuffer = new List<LfuNode<K, V>>(deleteBufferSize);

private void TryDeleteBufferedItems()
{
if (deleteBufferQueue.TryTake(out var buffer) == BufferStatus.Success)
{
foreach (var n in buffer)
{
this.dictionary.TryRemove(n.Key, out var _);
Disposer<V>.Dispose(n.Value);
}

buffer.Clear();
this.cleanBufferQueue.TryAdd(buffer);
}
}

private void DeleteAllBufferedItems()
{
var spinner = new SpinWait();
while (deleteBufferQueue.TryTake(out var buffer) != BufferStatus.Empty)
{
if (buffer != null)
{
foreach (var n in buffer)
{
this.dictionary.TryRemove(n.Key, out var _);
Disposer<V>.Dispose(n.Value);
}

buffer.Clear();
this.cleanBufferQueue.TryAdd(buffer);
}

spinner.SpinOnce();
}
}

private void OnAccess(LfuNode<K, V> node)
{
// there was a cache hit even if the item was removed or is not yet added.
Expand Down Expand Up @@ -709,9 +876,10 @@ private bool AdmitCandidate(K candidateKey, K victimKey)

private void Evict(LfuNode<K, V> evictee)
{
this.dictionary.TryRemove(evictee.Key, out var _);
//this.dictionary.TryRemove(evictee.Key, out var _);
TryAddToRemoveList(evictee);
evictee.list.Remove(evictee);
Disposer<V>.Dispose(evictee.Value);
// Disposer<V>.Dispose(evictee.Value);
this.metrics.evictedCount++;
}

Expand Down