buffer

BitFaster.Caching.Benchmarks/Lfu/LfuJustGetOrAdd.cs

@@ -22,11 +22,11 @@ public class LfuJustGetOrAdd
 
         const int stripes = 1;
         private static readonly BackgroundThreadScheduler background = new BackgroundThreadScheduler();
-        private static readonly ConcurrentLfu<int, int> concurrentLfu = new ConcurrentLfu<int, int>(stripes, 9, background, EqualityComparer<int>.Default);
+        private static readonly ConcurrentLfu<int, int> concurrentLfu = new ConcurrentLfu<int, int>(stripes, 9, background, EqualityComparer<int>.Default, LfuBufferSize.Default(1, 128));
 
-        private static readonly ConcurrentLfu<int, int> concurrentLfuFore = new ConcurrentLfu<int, int>(stripes, 9, new ForegroundScheduler(), EqualityComparer<int>.Default);
-        private static readonly ConcurrentLfu<int, int> concurrentLfuTp = new ConcurrentLfu<int, int>(stripes, 9, new ThreadPoolScheduler(), EqualityComparer<int>.Default);
-        private static readonly ConcurrentLfu<int, int> concurrentLfuNull = new ConcurrentLfu<int, int>(stripes, 9, new NullScheduler(), EqualityComparer<int>.Default);
+        private static readonly ConcurrentLfu<int, int> concurrentLfuFore = new ConcurrentLfu<int, int>(stripes, 9, new ForegroundScheduler(), EqualityComparer<int>.Default, LfuBufferSize.Default(1, 128));
+        private static readonly ConcurrentLfu<int, int> concurrentLfuTp = new ConcurrentLfu<int, int>(stripes, 9, new ThreadPoolScheduler(), EqualityComparer<int>.Default, LfuBufferSize.Default(1, 128));
+        private static readonly ConcurrentLfu<int, int> concurrentLfuNull = new ConcurrentLfu<int, int>(stripes, 9, new NullScheduler(), EqualityComparer<int>.Default, LfuBufferSize.Default(1, 128));
 
         [GlobalSetup]
         public void GlobalSetup()

BitFaster.Caching.Benchmarks/Lru/LruJustGetOrAdd.cs

@@ -48,7 +48,7 @@ public class LruJustGetOrAdd
         private static readonly ICache<int, int> atomicFastLru = new ConcurrentLruBuilder<int, int>().WithConcurrencyLevel(8).WithCapacity(9).WithAtomicGetOrAdd().Build();
 
         private static readonly BackgroundThreadScheduler background = new BackgroundThreadScheduler();
-        private static readonly ConcurrentLfu<int, int> concurrentLfu = new ConcurrentLfu<int, int>(1, 9, background, EqualityComparer<int>.Default);
+        private static readonly ConcurrentLfu<int, int> concurrentLfu = new ConcurrentLfu<int, int>(1, 9, background, EqualityComparer<int>.Default, LfuBufferSize.Default(1, 128));
 
 
         private static readonly int key = 1;

BitFaster.Caching.HitRateAnalysis/Analysis.cs

@@ -22,7 +22,7 @@ public Analysis(int cacheSize)
         {
             concurrentLru = new ConcurrentLru<K, int>(1, cacheSize, EqualityComparer<K>.Default);
             classicLru = new ClassicLru<K, int>(1, cacheSize, EqualityComparer<K>.Default);
-            concurrentLfu = new ConcurrentLfu<K, int>(1, cacheSize, new ForegroundScheduler(), EqualityComparer<K>.Default);
+            concurrentLfu = new ConcurrentLfu<K, int>(1, cacheSize, new ForegroundScheduler(), EqualityComparer<K>.Default, LfuBufferSize.Default(1, 128));
         }
 
         public int CacheSize => concurrentLru.Capacity;

BitFaster.Caching.ThroughputAnalysis/Program.cs

@@ -22,7 +22,7 @@ class Program
         const double s = 0.86;
         const int n = 500;
         static int capacity = 500;
-        const int maxThreads = 52;
+        const int maxThreads = 64;
         const int sampleCount = 2000;
         const int repeatCount = 400;
 
@@ -102,7 +102,7 @@ static void Main(string[] args)
                 for (int i = 0; i < warmup + runs; i++)
                 {
                     var scheduler = new BackgroundThreadScheduler();
-                    results[i] = MeasureThroughput(new ConcurrentLfu<int, int>(concurrencyLevel: tc, capacity: capacity, scheduler: scheduler, EqualityComparer<int>.Default), tc);
+                    results[i] = MeasureThroughput(new ConcurrentLfu<int, int>(concurrencyLevel: tc, capacity: capacity, scheduler: scheduler, EqualityComparer<int>.Default, LfuBufferSize.Default(concurrencyLevel: tc, capacity: capacity)), tc);
                     scheduler.Dispose();
                 }
                 avg = AverageLast(results, runs) / 1000000;

BitFaster.Caching.UnitTests/AssemblyInitialize.cs

@@ -1,9 +1,4 @@
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading;
-using System.Threading.Tasks;
+using System.Threading;
 using Xunit.Abstractions;
 using Xunit.Sdk;
 

BitFaster.Caching.UnitTests/BitFaster.Caching.UnitTests.csproj

@@ -17,6 +17,7 @@
       <PrivateAssets>all</PrivateAssets>
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
     </PackageReference>
+    <PackageReference Include="Xunit.SkippableFact" Version="1.4.13" />
   </ItemGroup>
 
   <ItemGroup>

BitFaster.Caching.UnitTests/Buffers/StripedBufferSizeTests.cs

@@ -0,0 +1,42 @@
+using System;
+using BitFaster.Caching.Buffers;
+using FluentAssertions;
+using Xunit;
+
+namespace BitFaster.Caching.UnitTests.Buffers
+{
+    public class StripedBufferSizeTests
+    {
+        [Fact]
+        public void WhenBufferSizeIsLessThan1CtorThrows()
+        {
+            Action constructor = () => { var x = new StripedBufferSize(-1, 1); };
+
+            constructor.Should().Throw<ArgumentOutOfRangeException>();
+        }
+
+        [Fact]
+        public void WhenStripeCountIsLessThan1CtorThrows()
+        {
+            Action constructor = () => { var x = new StripedBufferSize(1, -1); };
+
+            constructor.Should().Throw<ArgumentOutOfRangeException>();
+        }
+
+        [Fact]
+        public void SizeIsRoundedToNextPowerOf2()
+        {
+            var bs = new StripedBufferSize(6, 16);
+
+            bs.BufferSize.Should().Be(8);
+        }
+
+        [Fact]
+        public void StripeCountIsRoundedToNextPowerOf2()
+        {
+            var bs = new StripedBufferSize(16, 6);
+
+            bs.StripeCount.Should().Be(8);
+        }
+    }
+}

BitFaster.Caching.UnitTests/Lfu/ConcurrentLfuBuilderTests.cs

@@ -1,9 +1,6 @@
 using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
 using BitFaster.Caching.Atomic;
+using BitFaster.Caching.Buffers;
 using BitFaster.Caching.Lfu;
 using BitFaster.Caching.Scheduler;
 using FluentAssertions;
@@ -56,6 +53,17 @@ public void TestComparer()
             lfu.TryGet("A", out var value).Should().BeTrue();
         }
 
+        [Fact]
+        public void TestBufferConfiguraiton()
+        {
+            ICache<string, int> lfu = new ConcurrentLfuBuilder<string, int>()
+               .WithBufferConfiguration(new LfuBufferSize(
+                    new StripedBufferSize(128, 2),
+                    new StripedBufferSize(128, 2)
+                   ))
+               .Build();
+        }
+
         // 1
         [Fact]
         public void WithScopedValues()

BitFaster.Caching.UnitTests/Lfu/ConcurrentLfuTests.cs

@@ -3,9 +3,8 @@
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Linq;
-using System.Text;
-using System.Threading;
 using System.Threading.Tasks;
+using BitFaster.Caching.Buffers;
 using BitFaster.Caching.Lfu;
 using BitFaster.Caching.Scheduler;
 using BitFaster.Caching.UnitTests.Lru;
@@ -19,7 +18,7 @@ public class ConcurrentLfuTests
     {
         private readonly ITestOutputHelper output;
 
-        private ConcurrentLfu<int, int> cache = new ConcurrentLfu<int, int>(1, 20, new BackgroundThreadScheduler(), EqualityComparer<int>.Default);
+        private ConcurrentLfu<int, int> cache = new ConcurrentLfu<int, int>(1, 20, new BackgroundThreadScheduler(), EqualityComparer<int>.Default, LfuBufferSize.Default(1, 128));
         private ValueFactory valueFactory = new ValueFactory();
 
         public ConcurrentLfuTests(ITestOutputHelper output)
@@ -76,7 +75,7 @@ public void WhenItemsAddedExceedsCapacityItemsAreDiscarded()
         [Fact]
         public void WhenItemIsEvictedItIsDisposed()
         {
-            var dcache = new ConcurrentLfu<int, DisposableItem>(1, 20, new BackgroundThreadScheduler(), EqualityComparer<int>.Default);
+            var dcache = new ConcurrentLfu<int, DisposableItem>(1, 20, new BackgroundThreadScheduler(), EqualityComparer<int>.Default, LfuBufferSize.Default(1, 128));
             var disposables = new DisposableItem[25];
 
             for (int i = 0; i < 25; i++)
@@ -306,7 +305,7 @@ public void WriteUpdatesProtectedLruOrder()
         [Fact]
         public void WhenHitRateChangesWindowSizeIsAdapted()
         {
-            cache = new ConcurrentLfu<int, int>(1, 20, new NullScheduler(), EqualityComparer<int>.Default);
+            cache = new ConcurrentLfu<int, int>(1, 20, new NullScheduler(), EqualityComparer<int>.Default, LfuBufferSize.Default(1, 128));
 
             // First completely fill the cache, push entries into protected
             for (int i = 0; i < 20; i++)
@@ -375,13 +374,13 @@ public void WhenHitRateChangesWindowSizeIsAdapted()
         public void ReadSchedulesMaintenanceWhenBufferIsFull()
         {
             var scheduler = new TestScheduler();
-            cache = new ConcurrentLfu<int, int>(1, 20, scheduler, EqualityComparer<int>.Default);
+            cache = new ConcurrentLfu<int, int>(1, 20, scheduler, EqualityComparer<int>.Default, LfuBufferSize.Default(1, 128));
 
             cache.GetOrAdd(1, k => k);
             scheduler.RunCount.Should().Be(1);
             cache.PendingMaintenance();
 
-            for (int i = 0; i < ConcurrentLfu<int, int>.BufferSize; i++)
+            for (int i = 0; i < LfuBufferSize.DefaultBufferSize; i++)
             {
                 scheduler.RunCount.Should().Be(1);
                 cache.GetOrAdd(1, k => k);
@@ -395,29 +394,31 @@ public void ReadSchedulesMaintenanceWhenBufferIsFull()
         [Fact]
         public void WhenReadBufferIsFullReadsAreDropped()
         {
-            int bufferSize = ConcurrentLfu<int, int>.BufferSize;
             var scheduler = new TestScheduler();
-            cache = new ConcurrentLfu<int, int>(1, 20, scheduler, EqualityComparer<int>.Default);
+            cache = new ConcurrentLfu<int, int>(1, 20, scheduler, EqualityComparer<int>.Default, LfuBufferSize.Default(1, 128));
 
             cache.GetOrAdd(1, k => k);
             scheduler.RunCount.Should().Be(1);
             cache.PendingMaintenance();
 
-            for (int i = 0; i < bufferSize * 2; i++)
+            for (int i = 0; i < LfuBufferSize.DefaultBufferSize * 2; i++)
             {
                 cache.GetOrAdd(1, k => k);
             }
 
             cache.PendingMaintenance();
 
-            cache.Metrics.Value.Hits.Should().Be(bufferSize);
+            cache.Metrics.Value.Hits.Should().Be(LfuBufferSize.DefaultBufferSize);
         }
 
         [Fact]
         public void WhenWriteBufferIsFullAddDoesMaintenance()
         {
+            var bufferSize = LfuBufferSize.DefaultBufferSize;
             var scheduler = new TestScheduler();
-            cache = new ConcurrentLfu<int, int>(1, ConcurrentLfu<int, int>.BufferSize * 2, scheduler, EqualityComparer<int>.Default);
+
+            var bufferConfig = new LfuBufferSize(new StripedBufferSize(bufferSize, 1), new StripedBufferSize(bufferSize, 1));
+            cache = new ConcurrentLfu<int, int>(1, bufferSize * 2, scheduler, EqualityComparer<int>.Default, bufferConfig);
 
             // add an item, flush write buffer
             cache.GetOrAdd(-1, k => k);
@@ -430,7 +431,7 @@ public void WhenWriteBufferIsFullAddDoesMaintenance()
 
             // add buffer size items, last iteration will invoke maintenance on the foreground since write
             // buffer is full and test scheduler did not do any work
-            for (int i = 0; i < ConcurrentLfu<int, int>.BufferSize; i++)
+            for (int i = 0; i < bufferSize; i++)
             {
                 scheduler.RunCount.Should().Be(2);
                 cache.GetOrAdd(i, k => k);
@@ -444,9 +445,10 @@ public void WhenWriteBufferIsFullAddDoesMaintenance()
         [Fact]
         public void WhenWriteBufferIsFullUpdatesAreDropped()
         {
-            int bufferSize = ConcurrentLfu<int, int>.BufferSize;
+            var bufferSize = LfuBufferSize.DefaultBufferSize;
             var scheduler = new TestScheduler();
-            cache = new ConcurrentLfu<int, int>(1, 20, scheduler, EqualityComparer<int>.Default);
+            var bufferConfig = new LfuBufferSize(new StripedBufferSize(bufferSize, 1), new StripedBufferSize(bufferSize, 1));
+            cache = new ConcurrentLfu<int, int>(1, 20, scheduler, EqualityComparer<int>.Default, bufferConfig);
 
             cache.GetOrAdd(1, k => k);
             scheduler.RunCount.Should().Be(1);
@@ -578,7 +580,7 @@ public void WhenItemIsRemovedItIsRemoved()
         [Fact]
         public void WhenItemIsRemovedItIsDisposed()
         {
-            var dcache = new ConcurrentLfu<int, DisposableItem>(1, 20, new BackgroundThreadScheduler(), EqualityComparer<int>.Default);
+            var dcache = new ConcurrentLfu<int, DisposableItem>(1, 20, new BackgroundThreadScheduler(), EqualityComparer<int>.Default, LfuBufferSize.Default(1, 128));
             var disposable = new DisposableItem();
 
             dcache.GetOrAdd(1, k => disposable);
@@ -667,7 +669,7 @@ public void TrimRemovesNItems()
         public void TrimWhileItemsInWriteBufferRemovesNItems()
         {
             // null scheduler == no maintenance, all writes fit in buffer
-            cache = new ConcurrentLfu<int, int>(1, 20, new NullScheduler(), EqualityComparer<int>.Default);
+            cache = new ConcurrentLfu<int, int>(1, 20, new NullScheduler(), EqualityComparer<int>.Default, LfuBufferSize.Default(1, 128));
 
             for (int i = 0; i < 25; i++)
             {
@@ -705,7 +707,7 @@ public void VerifyHitsWithBackgroundScheduler()
         public void VerifyHitsWithThreadPoolScheduler()
         {
             // when running all tests in parallel, sample count drops significantly: set low bar for stability.
-            cache = new ConcurrentLfu<int, int>(1, 20, new ThreadPoolScheduler(), EqualityComparer<int>.Default);
+            cache = new ConcurrentLfu<int, int>(1, 20, new ThreadPoolScheduler(), EqualityComparer<int>.Default, LfuBufferSize.Default(1, 128));
             VerifyHits(iterations: 10000000, minSamples: 500000);
         }
 
@@ -715,7 +717,7 @@ public void VerifyHitsWithThreadPoolScheduler()
         [Fact]
         public void VerifyHitsWithNullScheduler()
         {
-            cache = new ConcurrentLfu<int, int>(1, 20, new NullScheduler(), EqualityComparer<int>.Default);
+            cache = new ConcurrentLfu<int, int>(1, 20, new NullScheduler(), EqualityComparer<int>.Default, LfuBufferSize.Default(1, 128));
             VerifyHits(iterations: 10000000, minSamples: -1);
         }
 
@@ -726,12 +728,12 @@ public void VerifyHitsWithNullScheduler()
         [Fact]
         public void VerifyHitsWithForegroundScheduler()
         {
-            cache = new ConcurrentLfu<int, int>(1, 20, new ForegroundScheduler(), EqualityComparer<int>.Default);
+            cache = new ConcurrentLfu<int, int>(1, 20, new ForegroundScheduler(), EqualityComparer<int>.Default, LfuBufferSize.Default(1, 128));
 
             // Note: TryAdd will drop 1 read per full read buffer, since TryAdd will return false
             // before TryScheduleDrain is called. This serves as sanity check.
             int iterations = 10000000;
-            int dropped = iterations / ConcurrentLfu<int, int>.BufferSize;
+            int dropped = iterations / LfuBufferSize.DefaultBufferSize;
 
             this.output.WriteLine($"Will drop {dropped} reads.");
 

BitFaster.Caching.UnitTests/Lfu/LfuBufferSizeTests.cs

@@ -0,0 +1,55 @@
+using System;
+using BitFaster.Caching.Buffers;
+using BitFaster.Caching.Lfu;
+using FluentAssertions;
+using Xunit;
+
+namespace BitFaster.Caching.UnitTests.Lfu
+{
+    public class LfuBufferSizeTests
+    {
+        [Fact]
+        public void WhenReadBufferIsNullThrows()
+        {
+            Action constructor = () => { var x = new LfuBufferSize(null, new StripedBufferSize(1, 1)); };
+
+            constructor.Should().Throw<ArgumentNullException>();
+        }
+
+        [Fact]
+        public void WhenWriteBufferIsNullThrows()
+        {
+            Action constructor = () => { var x = new LfuBufferSize(new StripedBufferSize(1, 1), null); };
+
+            constructor.Should().Throw<ArgumentNullException>();
+        }
+
+        [SkippableTheory]
+        [InlineData(1, 3, 1, 32, 1, 16)]
+        [InlineData(1, 14, 1, 128, 1, 16)]
+        [InlineData(1, 50, 1, 128, 1, 64)]
+        [InlineData(1, 100, 1, 128, 1, 128)]
+        [InlineData(4, 100, 4, 128, 4, 32)]
+        [InlineData(16, 100, 8, 128, 8, 16)] // fails win
+        [InlineData(64, 100, 8, 128, 8, 16)] // fails win
+        [InlineData(1, 1000, 1, 128, 1, 128)]
+        [InlineData(4, 1000, 4, 128, 4, 128)]
+        [InlineData(32, 1000, 32, 128, 32, 32)] // fails win + fails mac
+        [InlineData(256, 100000, 32, 128, 32, 32)] // fails win + fails mac
+        public void CalculateDefaultBufferSize(int concurrencyLevel, int capacity, int expectedReadStripes, int expectedReadBuffer, int expecteWriteStripes, int expecteWriteBuffer)
+        {
+            // Some of these tests depend on the CPU Core count - skip if run on a different config machine.
+            bool notExpectedCpuCount = Environment.ProcessorCount != 12;
+            bool concurrencyLevelThresholdExceeded = BitOps.CeilingPowerOfTwo(concurrencyLevel) > BitOps.CeilingPowerOfTwo(Environment.ProcessorCount * 2);
+
+            Skip.If(concurrencyLevelThresholdExceeded && notExpectedCpuCount, "Test outcome depends on machine CPU count");
+
+            var bufferSize = LfuBufferSize.Default(concurrencyLevel, capacity);
+
+            bufferSize.Read.StripeCount.Should().Be(expectedReadStripes);
+            bufferSize.Read.BufferSize.Should().Be(expectedReadBuffer);
+            bufferSize.Write.StripeCount.Should().Be(expecteWriteStripes);
+            bufferSize.Write.BufferSize.Should().Be(expecteWriteBuffer);
+        }
+    }
+}

BitFaster.Caching.UnitTests/Scheduler/BackgroundSchedulerTests.cs

@@ -69,23 +69,18 @@ public async Task WhenWorkThrowsLastExceptionIsPopulated()
         }
 
         [Fact]
-        public void WhenBacklogExceededThrows()
+        public void WhenBacklogExceededTasksAreDropped()
         {
             TaskCompletionSource tcs = new TaskCompletionSource();
 
-            Action start = () => 
+            for (int i = 0; i < BackgroundThreadScheduler.MaxBacklog * 2; i++)
             {
-                // Add 2 because 1 thread *may* be released, start running and then block before we attempt to schedule all tasks.
-                // this leaves BackgroundThreadScheduler.MaxBacklog slots available. So we need + 2 to guarantee all slots are
-                // used.
-                for (int i = 0; i < BackgroundThreadScheduler.MaxBacklog + 2; i++)
-                {
-                    scheduler.Run(() => { tcs.Task.Wait(); });
-                }
-            };
-
-            start.Should().Throw<InvalidOperationException>();
+                scheduler.Run(() => { tcs.Task.Wait(); });
+            }
+
             tcs.SetResult();
+
+            scheduler.RunCount.Should().BeCloseTo(BackgroundThreadScheduler.MaxBacklog, 1);
         }
 
         [Fact]