| | | 1 | | using AmbientServices.Utilities; |
| | | 2 | | using System; |
| | | 3 | | using System.Collections.Concurrent; |
| | | 4 | | using System.Collections.Generic; |
| | | 5 | | using System.Threading; |
| | | 6 | | using System.Threading.Tasks; |
| | | 7 | | |
| | | 8 | | namespace AmbientServices; |
| | | 9 | | |
| | | 10 | | /// <summary> |
| | | 11 | | /// A basic default implementation of <see cref="IAmbientLocalCache"/> providing a small, bounded, in-process cache. |
| | | 12 | | /// </summary> |
| | | 13 | | /// <remarks> |
| | | 14 | | /// <pitch>The zero-configuration local cache used unless overridden: a small in-process store with bounded bookkeeping, |
| | | 15 | | /// <pledge><see cref="IAmbientLocalCache"/></pledge> |
| | | 16 | | /// <pledge>A discarded entry is disposed through <see cref="IAsyncDisposable"/> when it implements that interface and t |
| | | 17 | | /// <plan> |
| | | 18 | | /// Entries live in a <see cref="ConcurrentDictionary{TKey,TValue}"/>, with two <see cref="ConcurrentQueue{T}"/>s of boo |
| | | 19 | | /// Every cache call increments an <see cref="Interlocked"/> counter, and ejection runs when that counter hits the confi |
| | | 20 | | /// Expiration comparisons use <see cref="AmbientClock"/> so tests control time deterministically. The ejection cadence |
| | | 21 | | /// Entries stored with dispose-on-discard are disposed on ejection, replacement, expiration, and clear, on a removal th |
| | | 22 | | /// Trade-offs: constant-time operations and no background threads, in exchange for approximate size bounds (queue count |
| | | 23 | | /// </plan> |
| | | 24 | | /// <priority> |
| | | 25 | | /// <see cref="IAmbientLocalCache"/> |
| | | 26 | | /// 1. Bounded, predictable per-call cost over accurate capacity management: ejection rides a call-count cadence over ap |
| | | 27 | | /// 2. No background threads over timely expiration: nothing sweeps until a later call happens to, so a cache that goes |
| | | 28 | | /// </priority> |
| | | 29 | | /// </remarks> |
| | | 30 | | [DefaultAmbientService] |
| | | 31 | | internal class BasicAmbientLocalCache : IAmbientLocalCache |
| | | 32 | | { |
| | | 33 | | /// <summary>Caps stale-queue draining per eject call so a pathological queue cannot spin unbounded in one async con |
| | | 34 | | private const int MaxEjectQueueDrainSteps = 65536; |
| | | 35 | | |
| | | 36 | | /// <summary>Maximum snapshot-and-eject passes in <see cref="Clear"/> before giving up on reaching an empty cache.</ |
| | | 37 | | private const int MaxClearPasses = 8; |
| | | 38 | | |
| | | 39 | | private static readonly AmbientService<IAmbientSettingsSet> _Settings = Ambient.GetService<IAmbientSettingsSet>(); |
| | | 40 | | |
| | | 41 | | private readonly IAmbientSetting<int> _callFrequencyToEject; |
| | | 42 | | private readonly IAmbientSetting<int> _countToEject; |
| | | 43 | | private readonly IAmbientSetting<int> _minCacheEntries; |
| | | 44 | | private int _expireCount; |
| | | 45 | | private ConcurrentQueue<TimedQueueEntry> _timedQueue = new(); // interlocked (make readonly when we no longer supp |
| | | 46 | | private ConcurrentQueue<string> _untimedQueue = new(); // interlocked (make readonly when we no longer supp |
| | | 47 | | private readonly ConcurrentDictionary<string, CacheEntry> _cache = new(); |
| | | 48 | | |
| | | 49 | | public BasicAmbientLocalCache() |
| | | 50 | | : this(_Settings.Local) |
| | | 51 | | { |
| | | 52 | | } |
| | | 53 | | |
| | | 54 | | public BasicAmbientLocalCache(IAmbientSettingsSet? settings) |
| | | 55 | | { |
| | | 56 | | _callFrequencyToEject = AmbientSettings.GetSetting<int>(settings, nameof(BasicAmbientLocalCache) + "-EjectFreque |
| | | 57 | | _countToEject = AmbientSettings.GetSetting<int>(settings, nameof(BasicAmbientLocalCache) + "-MaximumItemCount", |
| | | 58 | | _minCacheEntries = AmbientSettings.GetSetting<int>(settings, nameof(BasicAmbientLocalCache) + "-MinimumItemCount |
| | | 59 | | } |
| | | 60 | | |
| | | 61 | | private struct TimedQueueEntry |
| | | 62 | | { |
| | | 63 | | public string Key; |
| | | 64 | | public DateTime Expiration; |
| | | 65 | | } |
| | | 66 | | |
| | | 67 | | private class CacheEntry |
| | | 68 | | { |
| | | 69 | | public bool DisposeWhenDiscarding { get; private set; } |
| | | 70 | | public string Key; |
| | | 71 | | public DateTime? Expiration; |
| | | 72 | | public object Entry; |
| | | 73 | | |
| | 2 | 74 | | public CacheEntry(string key, DateTime? expiration, object entry, bool disposeWhenDiscarding) |
| | | 75 | | { |
| | 2 | 76 | | DisposeWhenDiscarding = disposeWhenDiscarding; |
| | 2 | 77 | | Key = key; |
| | 2 | 78 | | Expiration = expiration; |
| | 2 | 79 | | Entry = entry; |
| | 2 | 80 | | } |
| | | 81 | | public async ValueTask Dispose() |
| | | 82 | | { |
| | | 83 | | // if the entry is disposable, dispose it after removing it |
| | 2 | 84 | | if (DisposeWhenDiscarding) await DisposeDiscardedItem(Entry); |
| | 2 | 85 | | } |
| | | 86 | | } |
| | | 87 | | |
| | | 88 | | /// <summary> |
| | | 89 | | /// Disposes an item the cache is discarding, preferring asynchronous disposal when the item implements both interfa |
| | | 90 | | /// </summary> |
| | | 91 | | private static async ValueTask DisposeDiscardedItem(object item) |
| | | 92 | | { |
| | | 93 | | if (item is IAsyncDisposable asyncDisposable) await asyncDisposable.DisposeAsync(); |
| | | 94 | | else if (item is IDisposable disposable) disposable.Dispose(); |
| | | 95 | | } |
| | | 96 | | public async ValueTask<T?> Retrieve<T>(string key, TimeSpan? refresh = null, CancellationToken cancel = default) whe |
| | | 97 | | { |
| | | 98 | | CacheEntry? entry; |
| | | 99 | | if (_cache.TryGetValue(key, out entry)) |
| | | 100 | | { |
| | | 101 | | DateTime now = AmbientClock.UtcNow; |
| | | 102 | | // refresh expiration? |
| | | 103 | | if (refresh != null) |
| | | 104 | | { |
| | | 105 | | // update the expiration time in the cache entry and add a NEW timed queue entry (we'll ignore the other |
| | | 106 | | DateTime newExpiration = now.Add(refresh.Value); |
| | | 107 | | entry.Expiration = newExpiration; |
| | | 108 | | _timedQueue.Enqueue(new TimedQueueEntry { Key = key, Expiration = newExpiration }); |
| | | 109 | | } |
| | | 110 | | await EjectIfNeeded(); |
| | | 111 | | // no expiration or NOT expired? return the item now |
| | | 112 | | if (!(entry.Expiration < now)) |
| | | 113 | | { |
| | | 114 | | // dispose-on-discard items are handed off to a single consumer, so remove (without disposing--ownership |
| | | 115 | | if (entry.DisposeWhenDiscarding) ((ICollection<KeyValuePair<string, CacheEntry>>)_cache).Remove(new KeyV |
| | | 116 | | return entry.Entry as T; |
| | | 117 | | } |
| | | 118 | | // else this item is expired so remove it from the cache |
| | | 119 | | await EjectEntry(entry, cancel); |
| | | 120 | | } |
| | | 121 | | else |
| | | 122 | | { |
| | | 123 | | await EjectIfNeeded(); |
| | | 124 | | } |
| | | 125 | | return null; |
| | | 126 | | } |
| | | 127 | | |
| | | 128 | | public async ValueTask Store<T>(string itemKey, T item, bool disposeWhenDiscarding, TimeSpan? maxCacheDuration = nul |
| | | 129 | | { |
| | | 130 | | // does this entry *not* expire in the past? |
| | | 131 | | if (!(maxCacheDuration < TimeSpan.FromTicks(0))) |
| | | 132 | | { |
| | | 133 | | DateTime? actualExpiration = null; |
| | | 134 | | DateTime now = AmbientClock.UtcNow; |
| | | 135 | | if (maxCacheDuration != null) actualExpiration = now.Add(maxCacheDuration.Value); |
| | | 136 | | if (expiration != null && expiration.Value.Kind == DateTimeKind.Local) expiration = expiration.Value.ToUnive |
| | | 137 | | if (expiration < actualExpiration) actualExpiration = expiration; |
| | | 138 | | CacheEntry entry = new(itemKey, actualExpiration, item, disposeWhenDiscarding); |
| | | 139 | | // ConcurrentDictionary has no async update delegate, and its update delegate may run more than |
| | | 140 | | // once under contention — so a disposing side effect there can double-dispose. Do the compare-and-swap |
| | | 141 | | // ourselves: this displaces exactly one entry, which we then dispose once, awaited, outside the dictionary. |
| | | 142 | | CacheEntry? displaced = null; |
| | | 143 | | while (true) |
| | | 144 | | { |
| | | 145 | | if (_cache.TryGetValue(itemKey, out CacheEntry? old)) |
| | | 146 | | { |
| | | 147 | | if (_cache.TryUpdate(itemKey, entry, old)) |
| | | 148 | | { |
| | | 149 | | displaced = old; |
| | | 150 | | break; |
| | | 151 | | } |
| | | 152 | | } |
| | | 153 | | else if (_cache.TryAdd(itemKey, entry)) |
| | | 154 | | { |
| | | 155 | | break; |
| | | 156 | | } |
| | | 157 | | // another writer changed the slot first; re-read and retry |
| | | 158 | | } |
| | | 159 | | if (displaced != null) await displaced.Dispose(); |
| | | 160 | | if (actualExpiration == null) |
| | | 161 | | { |
| | | 162 | | _untimedQueue.Enqueue(itemKey); |
| | | 163 | | } |
| | | 164 | | else |
| | | 165 | | { |
| | | 166 | | _timedQueue.Enqueue(new TimedQueueEntry { Key = itemKey, Expiration = actualExpiration.Value }); |
| | | 167 | | } |
| | | 168 | | } |
| | | 169 | | else |
| | | 170 | | { |
| | | 171 | | // else this item is expired so dispose of it as if we had put it into the cache and then it expired (when t |
| | | 172 | | if (disposeWhenDiscarding) await DisposeDiscardedItem(item); |
| | | 173 | | } |
| | | 174 | | await EjectIfNeeded(); |
| | | 175 | | } |
| | | 176 | | |
| | | 177 | | private async ValueTask EjectIfNeeded() |
| | | 178 | | { |
| | | 179 | | int callFrequencyToEject = _callFrequencyToEject.Value; |
| | | 180 | | if (callFrequencyToEject <= 0) |
| | | 181 | | callFrequencyToEject = 1; |
| | | 182 | | |
| | | 183 | | int countToEject = _countToEject.Value; |
| | | 184 | | int opSerial = Interlocked.Increment(ref _expireCount); |
| | | 185 | | bool onCadence = (opSerial % callFrequencyToEject) == 0; |
| | | 186 | | int queueSum = _untimedQueue.Count + _timedQueue.Count; |
| | | 187 | | bool overCapacity = queueSum > countToEject; |
| | | 188 | | if (!onCadence && !overCapacity) |
| | | 189 | | return; |
| | | 190 | | |
| | | 191 | | int maxRounds = Math.Max(32, Math.Min(131072, 4 + 2 * Math.Max(queueSum, countToEject + 1))); |
| | | 192 | | for (int round = 0; round < maxRounds; round++) |
| | | 193 | | { |
| | | 194 | | await EjectOneTimed(); |
| | | 195 | | await EjectOneUntimed(); |
| | | 196 | | |
| | | 197 | | queueSum = _untimedQueue.Count + _timedQueue.Count; |
| | | 198 | | if (queueSum <= countToEject) |
| | | 199 | | break; |
| | | 200 | | } |
| | | 201 | | } |
| | | 202 | | |
| | | 203 | | private async ValueTask EjectOneTimed(CancellationToken cancel = default) |
| | | 204 | | { |
| | | 205 | | // have we hit the minimum number of items? |
| | | 206 | | if (_timedQueue.Count <= _minCacheEntries.Value) return; |
| | | 207 | | // removing at least one timed item (as well as any expired items we come across) |
| | | 208 | | bool unexpiredItemEjected = false; |
| | | 209 | | int steps = 0; |
| | | 210 | | while (steps++ < MaxEjectQueueDrainSteps && _timedQueue.TryDequeue(out TimedQueueEntry qEntry)) |
| | | 211 | | { |
| | | 212 | | // can we find this item in the cache? |
| | | 213 | | CacheEntry? entry; |
| | | 214 | | if (_cache.TryGetValue(qEntry.Key, out entry)) |
| | | 215 | | { |
| | | 216 | | // is the expiration still the same? |
| | | 217 | | if (qEntry.Expiration == entry.Expiration) |
| | | 218 | | { |
| | | 219 | | // remove it from the cache, even though it may not have expired yet because it's time to eject some |
| | | 220 | | await EjectEntry(entry, cancel); |
| | | 221 | | // fall through and check to see if the next item is already expired |
| | | 222 | | unexpiredItemEjected = true; |
| | | 223 | | } |
| | | 224 | | // the item was refreshed, so we should ignore this entry-- if we have already ejected an unexpired item |
| | | 225 | | else if (!unexpiredItemEjected) |
| | | 226 | | { |
| | | 227 | | continue; |
| | | 228 | | } |
| | | 229 | | } |
| | | 230 | | // else we couldn't find the entry in the cache, so just move to the next entry (unless we've already ejecte |
| | | 231 | | else if (!unexpiredItemEjected) |
| | | 232 | | { |
| | | 233 | | continue; |
| | | 234 | | } |
| | | 235 | | // peek at the next entry |
| | | 236 | | if (_timedQueue.TryPeek(out qEntry)) |
| | | 237 | | { |
| | | 238 | | // has this entry expired? continue looping so that we remove this one too, even though we didn't *have* |
| | | 239 | | if (qEntry.Expiration < AmbientClock.UtcNow) continue; |
| | | 240 | | // else the entry hasn't expired and we either removed an entry above or skipped this code, so we can ju |
| | | 241 | | } |
| | | 242 | | // if we get here, there is no reason to look at another timed entry |
| | | 243 | | break; |
| | | 244 | | } |
| | | 245 | | } |
| | | 246 | | |
| | | 247 | | private async ValueTask EjectOneUntimed(CancellationToken cancel = default) |
| | | 248 | | { |
| | | 249 | | // have we hit the minimum number of items? |
| | | 250 | | if (_untimedQueue.Count <= _minCacheEntries.Value) return; |
| | | 251 | | // remove one untimed entry |
| | | 252 | | int steps = 0; |
| | | 253 | | while (steps++ < MaxEjectQueueDrainSteps && _untimedQueue.TryDequeue(out string? key)) |
| | | 254 | | { |
| | | 255 | | // can we find this item in the cache? |
| | | 256 | | CacheEntry? entry; |
| | | 257 | | if (_cache.TryGetValue(key, out entry)) |
| | | 258 | | { |
| | | 259 | | // is the expiration still the same (ie. untimed)? |
| | | 260 | | if (entry.Expiration == null) |
| | | 261 | | { |
| | | 262 | | // remove it from the cache |
| | | 263 | | await EjectEntry(entry, cancel); |
| | | 264 | | // fall through and stop looping |
| | | 265 | | } |
| | | 266 | | else // else the item was refreshed, so we should ignore this entry and go around again to remove anothe |
| | | 267 | | { |
| | | 268 | | continue; |
| | | 269 | | } |
| | | 270 | | } |
| | | 271 | | // else we couldn't find the entry in the cache, so just move to the next entry |
| | | 272 | | else |
| | | 273 | | { |
| | | 274 | | continue; |
| | | 275 | | } |
| | | 276 | | // if we get here, there is no reason to look at another untimed entry |
| | | 277 | | break; |
| | | 278 | | } |
| | | 279 | | } |
| | | 280 | | |
| | | 281 | | public async ValueTask<T?> Remove<T>(string itemKey, CancellationToken cancel = default) where T : class |
| | | 282 | | { |
| | | 283 | | CacheEntry? disposeEntry; |
| | | 284 | | if (_cache.TryRemove(itemKey, out disposeEntry)) |
| | | 285 | | { |
| | | 286 | | // hand the item off to the caller if it's the requested type, transferring dispose responsibility |
| | | 287 | | if (disposeEntry.Entry is T) return (T?)disposeEntry.Entry; |
| | | 288 | | // otherwise the caller can't take ownership, so the cache disposes the discarded entry (no-op unless dispos |
| | | 289 | | await disposeEntry.Dispose(); |
| | | 290 | | } |
| | | 291 | | return default; |
| | | 292 | | } |
| | | 293 | | |
| | | 294 | | private async ValueTask EjectEntry(CacheEntry entry, CancellationToken cancel = default) |
| | | 295 | | { |
| | | 296 | | CacheEntry? disposeEntry; |
| | | 297 | | // race to remove the item from the cache--did we win the race? |
| | | 298 | | if (_cache.TryRemove(entry.Key, out disposeEntry)) |
| | | 299 | | { |
| | | 300 | | await disposeEntry!.Dispose(); // if it was successfully removed, it can't be null |
| | | 301 | | } |
| | | 302 | | } |
| | | 303 | | |
| | | 304 | | public async ValueTask Clear(CancellationToken cancel = default) |
| | | 305 | | { |
| | | 306 | | Interlocked.Exchange(ref _untimedQueue, new ConcurrentQueue<string>()); |
| | | 307 | | Interlocked.Exchange(ref _timedQueue, new ConcurrentQueue<TimedQueueEntry>()); |
| | | 308 | | |
| | | 309 | | for (int pass = 0; pass < MaxClearPasses; pass++) |
| | | 310 | | { |
| | | 311 | | KeyValuePair<string, CacheEntry>[] snapshot = _cache.ToArray(); |
| | | 312 | | if (snapshot.Length == 0) |
| | | 313 | | break; |
| | | 314 | | |
| | | 315 | | foreach (KeyValuePair<string, CacheEntry> kv in snapshot) |
| | | 316 | | { |
| | | 317 | | cancel.ThrowIfCancellationRequested(); |
| | | 318 | | await EjectEntry(kv.Value, cancel); |
| | | 319 | | } |
| | | 320 | | |
| | | 321 | | if (_cache.IsEmpty) |
| | | 322 | | break; |
| | | 323 | | } |
| | | 324 | | } |
| | | 325 | | } |