I've had some fun writing a sequence based cache that is going to be used to maintain a consistent stream withstanding network re-connections.
The idea is that subscribers of a network based stream will receive a sequence number with every value in the stream, which can be sent when connecting to the stream which will cause the server to re-send values after the given sequence number, from the cache.
The cache is only required to hold values for a given duration, which will provide a predictable 'maximum-loss' of values after a given time; while not dropping values in periods of high throughput, unlike some sort of circular buffer.
In my implementation, I only process and remove expired values when values are put into the cache - avoiding some sort of schedule based 'clean up' which I didn't really know where to start in terms of implementation, but would likely be useful.
I would like a second opinion on how I've attempted to implement this, and some ways this could be improved or how this problem could be re-thought out entirely.
/// <summary>
/// Stream cache that uses sequence numbers
/// </summary>
public class StreamSequenceCache<T> where T: class
{
/// <summary>
/// The time a cached entry should be kept for
/// </summary>
public TimeSpan TTL { get; set; }
/// <summary>
/// Used to inject date times to use when testing - defaults to DateTime.Now|UtcNow
/// </summary>
public IDateTimeAccessor DateTimeAccessor { get; set; } = SharedLibrary.DateTimeAccessor.Default;
/// <summary>
/// Used to make sure sequence numbers only increase, keeping the values array ordered by sequence numbers
/// </summary>
long lastSequenceNumber = long.MinValue;
/// <summary>
/// Values in the cache
/// </summary>
List<CacheEntry> values = new List<CacheEntry>();
/// <summary>
/// Put a value into the cache, for a given sequence number
/// </summary>
/// <param name="sequenceNumber"></param>
/// <param name="value"></param>
public void Put(long sequenceNumber, T value)
{
if (sequenceNumber <= lastSequenceNumber)
throw new InvalidOperationException("Sequence number must only be incremented");
var now = DateTimeAccessor.UtcNow;
values.Add(new CacheEntry(sequenceNumber, now, value));
lastSequenceNumber = sequenceNumber;
// Find the first non-expired index - which will be the count to remove all expired values
// preceding it
// The max is to turn -1 (not found) into 0 (remove none)
// Note that -1 could also mean all entries have expired, but that isn't possible since we just inserted a non-expired entry
var removeCount = Math.Max(values.FindIndex(e => e.Timestamp + TTL > now), 0);
values.RemoveRange(0, removeCount);
}
/// <summary>
/// Get the values after a given sequence number
/// </summary>
/// <param name="sequenceNumber"></param>
/// <returns></returns>
public IEnumerable<T> Get(long afterSequenceNumber)
{
try
{
// Find the sequence number to return values after
int index = values.BinarySearch(new CacheEntry(afterSequenceNumber));
// If the index is less than 0, then it is some complementary something I don't understand... but using the magic
// ~ operator I can get back the index it 'would have' been inserted, which is effectively the count to skip
if (index < 0)
index = ~index;
else
// Increment by 1, since we need to skip the value at the index
index++;
// Return values
return values.Skip(index)
.Select(s => s.Value);
}
catch(InvalidOperationException e)
{
// Sequence number not found. We simply ignore it, since it is probably to old to care.
return new List<T>();
}
}
/// <summary>
/// Container to hold a cached entry
/// </summary>
class CacheEntry : IComparable<CacheEntry>
{
/// <summary>
/// Used when binary searching for a sequence number
/// </summary>
/// <param name="sequenceNumber"></param>
public CacheEntry(long sequenceNumber)
{
SequenceNumber = sequenceNumber;
}
public CacheEntry(long sequenceNumber, DateTime timestamp, T value)
{
SequenceNumber = sequenceNumber;
Timestamp = timestamp;
Value = value;
}
public long SequenceNumber { get; set; }
public DateTime Timestamp { get; set; }
public T Value { get; set; }
/// <summary>
/// Compare only the sequence number. This is used when binary searching to find the entries to respond
/// to a get request for
/// </summary>
/// <param name="other"></param>
/// <returns></returns>
public int CompareTo(CacheEntry other)
{
return SequenceNumber.CompareTo(other.SequenceNumber);
}
}
}
Additionally, I have some unit tests which I used to test this class - which some feedback would be very valuable to someone who doesn't write many unit tests:
using Microsoft.VisualStudio.TestTools.UnitTesting;
using SharedLibrary.Caching;
using System;
using System.Linq;
using TestSharedLibary.Mocks;
namespace TestSharedLibary
{
[TestClass]
public class TestStreamSequenceCache
{
MockDateTimeAccessor dateTimeAccessor = new MockDateTimeAccessor();
[TestMethod, Description("Test putting values")]
public void TestPut()
{
var cache = new StreamSequenceCache<string>();
cache.DateTimeAccessor = dateTimeAccessor;
cache.Put(-1, "item 1");
cache.Put(0, "item 2");
cache.Put(1, "item 3");
cache.Put(2, "item 4");
// Make sure dublicate sequence throws an exception
Assert.ThrowsException<InvalidOperationException>(() =>
{
cache.Put(2, "item 5");
});
}
[TestMethod, Description("Test getting sequences from the cache")]
public void TestGet()
{
var cache = new StreamSequenceCache<string>();
cache.DateTimeAccessor = dateTimeAccessor;
cache.Put(0, "item 1");
cache.Put(1, "item 2");
cache.Put(2, "item 3");
cache.Put(3, "item 4");
cache.Put(4, "item 5");
Assert.AreEqual(0, cache.Get(5).Count());
Assert.AreEqual(5, cache.Get(-10).Count());
Assert.AreEqual(2, cache.Get(2).Count());
Assert.AreEqual(0, cache.Get(4).Count());
cache = new StreamSequenceCache<string>();
cache.DateTimeAccessor = dateTimeAccessor;
cache.Put(-5, "item 1");
cache.Put(-4, "item 2");
cache.Put(3, "item 3");
cache.Put(4, "item 4");
cache.Put(5, "item 5");
Assert.AreEqual(4, cache.Get(-5).Count());
Assert.AreEqual(3, cache.Get(-4).Count());
Assert.AreEqual(1, cache.Get(4).Count());
Assert.AreEqual(0, cache.Get(5).Count());
Assert.AreEqual(0, cache.Get(6).Count());
}
[TestMethod, Description("Test cache expiry")]
public void TestExpiry()
{
var cache = new StreamSequenceCache<string>();
cache.TTL = TimeSpan.FromSeconds(5);
cache.DateTimeAccessor = dateTimeAccessor;
var reference = DateTime.UtcNow;
dateTimeAccessor.UtcNow = reference.AddSeconds(0);
cache.Put(0, "item 1");
dateTimeAccessor.UtcNow = reference.AddSeconds(1);
cache.Put(1, "item 2");
dateTimeAccessor.UtcNow = reference.AddSeconds(2);
cache.Put(2, "item 3");
dateTimeAccessor.UtcNow = reference.AddSeconds(3);
cache.Put(3, "item 4");
dateTimeAccessor.UtcNow = reference.AddSeconds(4);
cache.Put(4, "item 5");
// Check all items were added
dateTimeAccessor.UtcNow = DateTime.UtcNow;
Assert.AreEqual(5, cache.Get(-100).Count());
// Check expiring 1 item
dateTimeAccessor.UtcNow = reference.AddSeconds(5.5);
cache.Put(5, "item 6");
// Should still be 5 items, since the last should have been dropped
Assert.AreEqual(5, cache.Get(-100).Count());
// All items should be dropped
dateTimeAccessor.UtcNow = reference.AddSeconds(20);
cache.Put(6, "item 7");
// Should still be 5 items, since the last should have been dropped
Assert.AreEqual(1, cache.Get(-100).Count());
}
}
}