GLPriceCache Dispose shoud exit quickly
This commit is contained in:
@@ -2,57 +2,66 @@
|
||||
using MarketData.Utils;
|
||||
using MarketData.DataAccess;
|
||||
using System.Collections.Concurrent;
|
||||
using System.Collections.Generic;
|
||||
using System;
|
||||
using System.Linq;
|
||||
using System.Collections.Generic;
|
||||
using System.Threading;
|
||||
using System.Linq;
|
||||
using System.Threading.Tasks;
|
||||
|
||||
namespace MarketData.Cache
|
||||
{
|
||||
/// <summary>
|
||||
/// GLPriceCache - Used by Gain/Loss Generator which are usedby MarketDataSErver and the User Interface.
|
||||
/// The entire cache evicts every hour so calling Add(PortfolioTrades portfolioTrades) will force price reload
|
||||
/// Always ensures that open trade symbols get most recent price from the database.
|
||||
/// </summary>
|
||||
public class GLPriceCache : IDisposable
|
||||
{
|
||||
// ── Singleton ────────────────────────────────────────────────────────────
|
||||
// -- Singleton ------------------------------------------------------------
|
||||
private static readonly object instanceLock = new object();
|
||||
private static GLPriceCache instance = null;
|
||||
|
||||
// -- Disposal -------------------------------------------------------------
|
||||
private volatile bool disposed = false;
|
||||
private int refreshInProgress = 0;
|
||||
|
||||
/// <summary>
|
||||
///
|
||||
/// </summary>
|
||||
/// <returns></returns>
|
||||
public static GLPriceCache GetInstance()
|
||||
{
|
||||
lock (instanceLock)
|
||||
{
|
||||
if (instance == null)
|
||||
instance = new GLPriceCache();
|
||||
if (instance == null)instance = new GLPriceCache();
|
||||
return instance;
|
||||
}
|
||||
}
|
||||
|
||||
// ── State ────────────────────────────────────────────────────────────────
|
||||
// -- State ----------------------------------------------------------------
|
||||
private readonly Dictionary<string, PricesByDate> priceCache = new Dictionary<string, PricesByDate>();
|
||||
private readonly ConcurrentDictionary<string, object> symbolFetchLocks = new ConcurrentDictionary<string, object>();
|
||||
private readonly object cacheLock = new object();
|
||||
private DateTime latestDate = Utility.Epoch;
|
||||
|
||||
// ── Background refresh ───────────────────────────────────────────────────
|
||||
// -- Background refresh ---------------------------------------------------
|
||||
private readonly TimeSpan cacheCycle = TimeSpan.FromMinutes(5);
|
||||
private Timer refreshTimer = null;
|
||||
private int tickCount = 0;
|
||||
private const int evictionTickInterval = 12; // every 12 ticks x 5 min = 1 hour
|
||||
|
||||
// ── Parallelism ──────────────────────────────────────────────────────────
|
||||
// -- Parallelism ----------------------------------------------------------
|
||||
private static readonly int maxParallelDbCalls = ResolveMaxParallelDbCalls();
|
||||
|
||||
private static int ResolveMaxParallelDbCalls()
|
||||
{
|
||||
int @default = Math.Min(Math.Max(1, Environment.ProcessorCount) * 3, 32);
|
||||
string configured = Environment.GetEnvironmentVariable("GL_PRICE_CACHE_PARALLEL_DB_CALLS");
|
||||
return int.TryParse(configured, out int parsed) && parsed > 0 ? parsed : @default;
|
||||
return Math.Min(Math.Max(1, Environment.ProcessorCount) * 3, 32);
|
||||
// int @default = Math.Min(Math.Max(1, Environment.ProcessorCount) * 3, 32);
|
||||
// string configured = Environment.GetEnvironmentVariable("GL_PRICE_CACHE_PARALLEL_DB_CALLS");
|
||||
// return int.TryParse(configured, out int parsed) && parsed > 0 ? parsed : @default;
|
||||
}
|
||||
|
||||
// ── Disposal ─────────────────────────────────────────────────────────────
|
||||
private volatile bool disposed = false;
|
||||
private int refreshInProgress = 0;
|
||||
|
||||
// ── Constructor ──────────────────────────────────────────────────────────
|
||||
// -- Constructor ----------------------------------------------------------
|
||||
private GLPriceCache()
|
||||
{
|
||||
refreshTimer = new Timer(OnCacheRefreshTick, null, cacheCycle, cacheCycle);
|
||||
@@ -71,7 +80,7 @@ namespace MarketData.Cache
|
||||
|
||||
if (disposing)
|
||||
{
|
||||
Timer timerToDispose;
|
||||
Timer timerToDispose = null;
|
||||
lock (instanceLock)
|
||||
{
|
||||
timerToDispose = refreshTimer;
|
||||
@@ -79,19 +88,49 @@ namespace MarketData.Cache
|
||||
instance = null;
|
||||
}
|
||||
|
||||
// Block until any in-flight tick completes before disposing
|
||||
using (var waited = new ManualResetEventSlim(false))
|
||||
{
|
||||
if (timerToDispose != null)
|
||||
timerToDispose.Dispose(waited.WaitHandle);
|
||||
waited.Wait(TimeSpan.FromSeconds(10));
|
||||
}
|
||||
// Dispose timer immediately; no blocking wait
|
||||
timerToDispose?.Dispose();
|
||||
|
||||
MDTrace.WriteLine(LogLevel.DEBUG, "[GLPriceCache:Dispose] Disposed.");
|
||||
}
|
||||
}
|
||||
|
||||
// ── Background tick ──────────────────────────────────────────────────────
|
||||
/// <summary>
|
||||
/// Dispoal
|
||||
/// </summary>
|
||||
//public void Dispose()
|
||||
//{
|
||||
// Dispose(true);
|
||||
// GC.SuppressFinalize(this);
|
||||
//}
|
||||
|
||||
//protected virtual void Dispose(bool disposing)
|
||||
//{
|
||||
// if (disposed) return;
|
||||
// disposed = true;
|
||||
|
||||
// if (disposing)
|
||||
// {
|
||||
// Timer timerToDispose;
|
||||
// lock (instanceLock)
|
||||
// {
|
||||
// timerToDispose = refreshTimer;
|
||||
// refreshTimer = null;
|
||||
// instance = null;
|
||||
// }
|
||||
// // Block until any in-flight tick completes before disposing
|
||||
// using (ManualResetEventSlim waited = new ManualResetEventSlim(false))
|
||||
// {
|
||||
// if (timerToDispose != null)
|
||||
// timerToDispose.Dispose(waited.WaitHandle);
|
||||
// waited.Wait(TimeSpan.FromSeconds(10));
|
||||
// }
|
||||
|
||||
// MDTrace.WriteLine(LogLevel.DEBUG, "[GLPriceCache:Dispose] Disposed.");
|
||||
// }
|
||||
//}
|
||||
|
||||
// -- Background tick ------------------------------------------------------
|
||||
private void OnCacheRefreshTick(object state)
|
||||
{
|
||||
if (disposed) return;
|
||||
@@ -100,30 +139,27 @@ namespace MarketData.Cache
|
||||
lock (cacheLock)
|
||||
{
|
||||
long count = CountInternal();
|
||||
MDTrace.WriteLine(LogLevel.DEBUG,
|
||||
$"[GLPriceCache:Tick] Symbols: {priceCache.Keys.Count}. " +
|
||||
$"Items in cache: {Utility.FormatNumber(count, 0, true)}.");
|
||||
MDTrace.WriteLine(LogLevel.DEBUG,$"[GLPriceCache:Tick] Symbols: {priceCache.Keys.Count}. " + $"Items in cache: {Utility.FormatNumber(count, 0, true)}.");
|
||||
}
|
||||
|
||||
if (++tickCount % evictionTickInterval == 0)
|
||||
{
|
||||
EvictStaleSymbols();
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
MDTrace.WriteLine(LogLevel.DEBUG,
|
||||
$"[GLPriceCache:Tick] [ERROR] Unhandled exception: {ex}");
|
||||
MDTrace.WriteLine(LogLevel.DEBUG,$"[GLPriceCache:Tick] [ERROR] Unhandled exception: {ex}");
|
||||
}
|
||||
}
|
||||
|
||||
// ── Eviction ─────────────────────────────────────────────────────────────
|
||||
// -- Eviction -------------------------------------------------------------
|
||||
private void EvictStaleSymbols()
|
||||
{
|
||||
if (Interlocked.CompareExchange(ref refreshInProgress, 1, 0) == 1)
|
||||
{
|
||||
MDTrace.WriteLine(LogLevel.DEBUG, "[GLPriceCache:Evict] Skipped — refresh in progress.");
|
||||
MDTrace.WriteLine(LogLevel.DEBUG, "[GLPriceCache:Evict] Skipped refresh in progress.");
|
||||
return;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
int count;
|
||||
@@ -133,9 +169,7 @@ namespace MarketData.Cache
|
||||
priceCache.Clear();
|
||||
symbolFetchLocks.Clear();
|
||||
}
|
||||
|
||||
MDTrace.WriteLine(LogLevel.DEBUG,
|
||||
$"[GLPriceCache:Evict] Cache cleared. {count} symbols evicted.");
|
||||
MDTrace.WriteLine(LogLevel.DEBUG,$"[GLPriceCache:Evict] Cache cleared. {count} symbols evicted.");
|
||||
}
|
||||
finally
|
||||
{
|
||||
@@ -143,89 +177,185 @@ namespace MarketData.Cache
|
||||
}
|
||||
}
|
||||
|
||||
// ── Public API ───────────────────────────────────────────────────────────
|
||||
// -- Public API -----------------------------------------------------------
|
||||
|
||||
// public void Add(PortfolioTrades portfolioTrades)
|
||||
// {
|
||||
// List<string> symbols = portfolioTrades.Symbols;
|
||||
|
||||
// Dictionary<string, DateTime> minTradeDates = symbols.ToDictionary(
|
||||
// sym => sym, sym => portfolioTrades.GetMinTradeDate(sym));
|
||||
|
||||
// // Only open positions need an intraday price refresh.
|
||||
// // Closed positions, regardless of close date, have immutable prices skip the DB call.
|
||||
// HashSet<string> mutableSymbols = new HashSet<string>(
|
||||
// symbols.Where(sym => portfolioTrades.HasOpenPositions(sym)));
|
||||
|
||||
// Dictionary<string, DateTime> minCacheDates;
|
||||
// lock (cacheLock)
|
||||
// {
|
||||
// minCacheDates = symbols.ToDictionary(
|
||||
// sym => sym,
|
||||
// sym => priceCache.ContainsKey(sym) ? priceCache[sym].MinDate : DateTime.MaxValue);
|
||||
// }
|
||||
|
||||
// ConcurrentDictionary<string, Prices> fetchedPrices = new ConcurrentDictionary<string, Prices>();
|
||||
// ConcurrentDictionary<string, Price> latestPrices = new ConcurrentDictionary<string, Price>();
|
||||
|
||||
// Parallel.ForEach(symbols, new ParallelOptions { MaxDegreeOfParallelism = maxParallelDbCalls }, symbol =>
|
||||
// {
|
||||
// if (disposed) return;
|
||||
|
||||
// DateTime minTradeDate = minTradeDates[symbol];
|
||||
// DateTime minCacheDate = minCacheDates[symbol];
|
||||
|
||||
// // Acquire a per-symbol lock to prevent concurrent clients from issuing
|
||||
// // duplicate DB fetches for the same symbol. First thread fetches;
|
||||
// // subsequent threads for the same symbol block here, then re-check
|
||||
// // the cache on entry and skip the fetch if already populated.
|
||||
// object symbolLock = symbolFetchLocks.GetOrAdd(symbol, _ => new object());
|
||||
// lock (symbolLock)
|
||||
// {
|
||||
// try
|
||||
// {
|
||||
// // Re-check cache state after acquiring symbol lock another
|
||||
// // client thread may have already fetched this symbol while we waited
|
||||
// lock (cacheLock)
|
||||
// {
|
||||
// if (priceCache.ContainsKey(symbol))
|
||||
// minCacheDate = priceCache[symbol].MinDate;
|
||||
// }
|
||||
|
||||
// // Historical fetch only when cache is missing or incomplete
|
||||
// Prices prices = null;
|
||||
// if (minCacheDate == DateTime.MaxValue)
|
||||
// prices = PricingDA.GetPrices(symbol, minTradeDate);
|
||||
// else if (minTradeDate < minCacheDate)
|
||||
// prices = PricingDA.GetPrices(symbol, minCacheDate, minTradeDate);
|
||||
|
||||
// if (prices != null && prices.Count > 0)
|
||||
// fetchedPrices[symbol] = prices;
|
||||
|
||||
// // Intraday refresh mutable symbols only
|
||||
// if (mutableSymbols.Contains(symbol))
|
||||
// {
|
||||
// Price latestPrice = PricingDA.GetPrice(symbol);
|
||||
// if (latestPrice != null)
|
||||
// latestPrices[symbol] = latestPrice;
|
||||
// }
|
||||
// }
|
||||
// catch (Exception ex)
|
||||
// {
|
||||
// MDTrace.WriteLine(LogLevel.DEBUG,
|
||||
// $"[GLPriceCache:Add] [ERROR] Failed fetching prices for {symbol}: {ex.Message}");
|
||||
// }
|
||||
// }
|
||||
// });
|
||||
|
||||
// lock (cacheLock)
|
||||
// {
|
||||
// // Historical prices idempotent, will not overwrite existing entries
|
||||
// foreach (KeyValuePair<string, Prices> kvp in fetchedPrices)
|
||||
// foreach (Price price in kvp.Value)
|
||||
// AddInternal(price);
|
||||
|
||||
// // Latest prices unconditional overwrite to capture intraday updates
|
||||
// foreach (KeyValuePair<string, Price> kvp in latestPrices)
|
||||
// {
|
||||
// PricesByDate pricesByDate;
|
||||
// if (!priceCache.TryGetValue(kvp.Key, out pricesByDate))
|
||||
// {
|
||||
// pricesByDate = new PricesByDate();
|
||||
// priceCache[kvp.Key] = pricesByDate;
|
||||
// }
|
||||
|
||||
// if (pricesByDate.ContainsKey(kvp.Value.Date))
|
||||
// pricesByDate.Remove(kvp.Value.Date);
|
||||
// pricesByDate.Add(kvp.Value.Date, kvp.Value);
|
||||
// }
|
||||
// }
|
||||
|
||||
// MDTrace.WriteLine(LogLevel.DEBUG,
|
||||
// $"[GLPriceCache:Add] Symbols: {symbols.Count}, Mutable: {mutableSymbols.Count}, " +
|
||||
// $"Historical fetches: {fetchedPrices.Count}, Intraday updates: {latestPrices.Count}");
|
||||
// }
|
||||
|
||||
public void Add(PortfolioTrades portfolioTrades)
|
||||
{
|
||||
List<string> symbols = portfolioTrades.Symbols;
|
||||
|
||||
Dictionary<string, DateTime> minTradeDates = symbols.ToDictionary(
|
||||
sym => sym, sym => portfolioTrades.GetMinTradeDate(sym));
|
||||
// Map each symbol to its earliest trade date
|
||||
Dictionary<string, DateTime> minTradeDates = symbols.ToDictionary(symbol => symbol, symbol => portfolioTrades.GetMinTradeDate(symbol));
|
||||
|
||||
// Only open positions need an intraday price refresh.
|
||||
// Closed positions, regardless of close date, have immutable prices — skip the DB call.
|
||||
HashSet<string> mutableSymbols = new HashSet<string>(
|
||||
symbols.Where(sym => portfolioTrades.HasOpenPositions(sym)));
|
||||
|
||||
Dictionary<string, DateTime> minCacheDates;
|
||||
// Only open positions need intraday price refresh
|
||||
HashSet<string> mutableSymbols = new HashSet<string>(symbols.Where(sym => portfolioTrades.HasOpenPositions(sym)));
|
||||
// Pre-filter symbols to skip fully cached immutable symbols
|
||||
List<string> symbolsToProcess = symbols
|
||||
.Where(sym =>
|
||||
{
|
||||
DateTime minCacheDate;
|
||||
lock (cacheLock)
|
||||
{
|
||||
minCacheDates = symbols.ToDictionary(
|
||||
sym => sym,
|
||||
sym => priceCache.ContainsKey(sym) ? priceCache[sym].MinDate : DateTime.MaxValue);
|
||||
minCacheDate = priceCache.ContainsKey(sym) ? priceCache[sym].MinDate : DateTime.MaxValue;
|
||||
}
|
||||
DateTime minTradeDate = minTradeDates[sym];
|
||||
// Skip if symbol is immutable AND cache already covers all trade dates
|
||||
return mutableSymbols.Contains(sym) || minTradeDate < minCacheDate;
|
||||
})
|
||||
.ToList();
|
||||
|
||||
if (symbolsToProcess.Count == 0) return; // nothing to do
|
||||
ConcurrentDictionary<string, Prices> fetchedPrices = new ConcurrentDictionary<string, Prices>();
|
||||
ConcurrentDictionary<string, Price> latestPrices = new ConcurrentDictionary<string, Price>();
|
||||
|
||||
Parallel.ForEach(symbols, new ParallelOptions { MaxDegreeOfParallelism = maxParallelDbCalls }, symbol =>
|
||||
Parallel.ForEach(symbolsToProcess, new ParallelOptions { MaxDegreeOfParallelism = maxParallelDbCalls }, symbol =>
|
||||
{
|
||||
if (disposed) return;
|
||||
|
||||
DateTime minTradeDate = minTradeDates[symbol];
|
||||
DateTime minCacheDate = minCacheDates[symbol];
|
||||
|
||||
// Acquire a per-symbol lock to prevent concurrent clients from issuing
|
||||
// duplicate DB fetches for the same symbol. First thread fetches;
|
||||
// subsequent threads for the same symbol block here, then re-check
|
||||
// the cache on entry and skip the fetch if already populated.
|
||||
DateTime minCacheDate;
|
||||
lock (cacheLock)
|
||||
{
|
||||
minCacheDate = priceCache.ContainsKey(symbol) ? priceCache[symbol].MinDate : DateTime.MaxValue;
|
||||
}
|
||||
object symbolLock = symbolFetchLocks.GetOrAdd(symbol, _ => new object());
|
||||
lock (symbolLock)
|
||||
{
|
||||
try
|
||||
{
|
||||
// Re-check cache state after acquiring symbol lock — another
|
||||
// client thread may have already fetched this symbol while we waited
|
||||
// Re-check cache after acquiring lock
|
||||
lock (cacheLock)
|
||||
{
|
||||
if (priceCache.ContainsKey(symbol))
|
||||
minCacheDate = priceCache[symbol].MinDate;
|
||||
if (priceCache.ContainsKey(symbol))minCacheDate = priceCache[symbol].MinDate;
|
||||
}
|
||||
|
||||
// Historical fetch — only when cache is missing or incomplete
|
||||
Prices prices = null;
|
||||
if (minCacheDate == DateTime.MaxValue)
|
||||
prices = PricingDA.GetPrices(symbol, minTradeDate);
|
||||
else if (minTradeDate < minCacheDate)
|
||||
prices = PricingDA.GetPrices(symbol, minCacheDate, minTradeDate);
|
||||
|
||||
if (prices != null && prices.Count > 0)
|
||||
fetchedPrices[symbol] = prices;
|
||||
|
||||
// Intraday refresh — mutable symbols only
|
||||
// Historical fetch only if cache is missing or incomplete
|
||||
if (minCacheDate == DateTime.MaxValue)prices = PricingDA.GetPrices(symbol, minTradeDate);
|
||||
else if (minTradeDate < minCacheDate)prices = PricingDA.GetPrices(symbol, minCacheDate, minTradeDate);
|
||||
if (prices != null && prices.Count > 0)fetchedPrices[symbol] = prices;
|
||||
// Intraday refresh for mutable symbols only
|
||||
if (mutableSymbols.Contains(symbol))
|
||||
{
|
||||
Price latestPrice = PricingDA.GetPrice(symbol);
|
||||
if (latestPrice != null)
|
||||
latestPrices[symbol] = latestPrice;
|
||||
if (latestPrice != null)latestPrices[symbol] = latestPrice;
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
MDTrace.WriteLine(LogLevel.DEBUG,
|
||||
$"[GLPriceCache:Add] [ERROR] Failed fetching prices for {symbol}: {ex.Message}");
|
||||
MDTrace.WriteLine(LogLevel.DEBUG,$"[GLPriceCache:Add] [ERROR] Failed fetching prices for {symbol}: {ex.Message}");
|
||||
}
|
||||
}
|
||||
});
|
||||
} // lcok(symLock)
|
||||
}); // Parallel
|
||||
|
||||
lock (cacheLock)
|
||||
{
|
||||
// Historical prices — idempotent, will not overwrite existing entries
|
||||
// Historical prices idempotent, do not overwrite existing entries
|
||||
foreach (KeyValuePair<string, Prices> kvp in fetchedPrices)
|
||||
{
|
||||
foreach (Price price in kvp.Value)
|
||||
{
|
||||
AddInternal(price);
|
||||
|
||||
// Latest prices — unconditional overwrite to capture intraday updates
|
||||
}
|
||||
}
|
||||
// Latest prices unconditional overwrite to capture intraday updates
|
||||
foreach (KeyValuePair<string, Price> kvp in latestPrices)
|
||||
{
|
||||
PricesByDate pricesByDate;
|
||||
@@ -234,39 +364,47 @@ namespace MarketData.Cache
|
||||
pricesByDate = new PricesByDate();
|
||||
priceCache[kvp.Key] = pricesByDate;
|
||||
}
|
||||
|
||||
if (pricesByDate.ContainsKey(kvp.Value.Date))
|
||||
{
|
||||
pricesByDate.Remove(kvp.Value.Date);
|
||||
}
|
||||
pricesByDate.Add(kvp.Value.Date, kvp.Value);
|
||||
}
|
||||
}
|
||||
|
||||
MDTrace.WriteLine(LogLevel.DEBUG,
|
||||
$"[GLPriceCache:Add] Symbols: {symbols.Count}, Mutable: {mutableSymbols.Count}, " +
|
||||
// Only log if we actually fetched or updated something
|
||||
if (fetchedPrices.Count > 0 || latestPrices.Count > 0)
|
||||
{
|
||||
MDTrace.WriteLine(LogLevel.DEBUG,$"[GLPriceCache:Add] Symbols processed: {symbolsToProcess.Count}, Mutable: {mutableSymbols.Count}, " +
|
||||
$"Historical fetches: {fetchedPrices.Count}, Intraday updates: {latestPrices.Count}");
|
||||
}
|
||||
|
||||
public void Add(Prices prices)
|
||||
{
|
||||
foreach (Price price in prices)
|
||||
Add(price);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Add prices
|
||||
/// </summary>
|
||||
/// <param name="prices"></param>
|
||||
public void Add(Prices prices)
|
||||
{
|
||||
foreach (Price price in prices)Add(price);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Add(List<string> symbols, DateTime pricingDate)
|
||||
/// </summary>
|
||||
/// <param name="symbols"></param>
|
||||
/// <param name="pricingDate"></param>
|
||||
public void Add(List<string> symbols, DateTime pricingDate)
|
||||
{
|
||||
if (symbols == null || symbols.Count == 0) return;
|
||||
|
||||
ConcurrentDictionary<string, Price> fetchedPrices = new ConcurrentDictionary<string, Price>();
|
||||
|
||||
Parallel.ForEach(symbols, new ParallelOptions { MaxDegreeOfParallelism = maxParallelDbCalls }, symbol =>
|
||||
{
|
||||
if (disposed) return;
|
||||
|
||||
lock (cacheLock)
|
||||
{
|
||||
if (ContainsPriceInternal(symbol, pricingDate)) return;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
Price price = PricingDA.GetPrice(symbol, pricingDate);
|
||||
@@ -274,18 +412,23 @@ namespace MarketData.Cache
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
MDTrace.WriteLine(LogLevel.DEBUG,
|
||||
$"[GLPriceCache:Add] [ERROR] Failed fetching price for {symbol} on {pricingDate:yyyy-MM-dd}: {ex.Message}");
|
||||
MDTrace.WriteLine(LogLevel.DEBUG,$"[GLPriceCache:Add] [ERROR] Failed fetching price for {symbol} on {pricingDate:yyyy-MM-dd}: {ex.Message}");
|
||||
}
|
||||
});
|
||||
|
||||
lock (cacheLock)
|
||||
{
|
||||
foreach (KeyValuePair<string, Price> kvp in fetchedPrices)
|
||||
{
|
||||
AddInternal(kvp.Value);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Add(Price price)
|
||||
/// </summary>
|
||||
/// <param name="price"></param>
|
||||
public void Add(Price price)
|
||||
{
|
||||
lock (cacheLock)
|
||||
@@ -294,14 +437,16 @@ namespace MarketData.Cache
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Refresh
|
||||
/// </summary>
|
||||
public void Refresh()
|
||||
{
|
||||
if (Interlocked.CompareExchange(ref refreshInProgress, 1, 0) == 1)
|
||||
{
|
||||
MDTrace.WriteLine(LogLevel.DEBUG, "[GLPriceCache:Refresh] Skipped — refresh already in progress.");
|
||||
MDTrace.WriteLine(LogLevel.DEBUG, "[GLPriceCache:Refresh] Skipped refresh already in progress.");
|
||||
return;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
List<string> symbols;
|
||||
@@ -312,23 +457,19 @@ namespace MarketData.Cache
|
||||
symbols = priceCache.Keys.ToList();
|
||||
currentMaxDates = priceCache.ToDictionary(kvp => kvp.Key, kvp => kvp.Value.MaxDate);
|
||||
}
|
||||
|
||||
if (symbols.Count == 0) return;
|
||||
|
||||
ConcurrentDictionary<string, PricesByDate> fullReloads = new ConcurrentDictionary<string, PricesByDate>();
|
||||
ConcurrentDictionary<string, Price> singleUpdates = new ConcurrentDictionary<string, Price>();
|
||||
|
||||
// Fetch outside the cache lock — no timeout risk holding cacheLock over I/O
|
||||
// Fetch outside the cache lock no timeout risk holding cacheLock over I/O
|
||||
Dictionary<string, DateTime> maxDbDates = PricingDA.GetLatestDates(symbols);
|
||||
DateTime latestDateFromDb = PricingDA.GetLatestDate();
|
||||
|
||||
Parallel.ForEach(symbols, new ParallelOptions { MaxDegreeOfParallelism = maxParallelDbCalls }, symbol =>
|
||||
{
|
||||
if (disposed) return;
|
||||
|
||||
DateTime cachedMax;
|
||||
if (!currentMaxDates.TryGetValue(symbol, out cachedMax)) return;
|
||||
|
||||
try
|
||||
{
|
||||
DateTime dbMax;
|
||||
@@ -345,40 +486,35 @@ namespace MarketData.Cache
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
MDTrace.WriteLine(LogLevel.DEBUG,
|
||||
$"[GLPriceCache:Refresh] [ERROR] Failed refreshing {symbol}: {ex.Message}");
|
||||
MDTrace.WriteLine(LogLevel.DEBUG,$"[GLPriceCache:Refresh] [ERROR] Failed refreshing {symbol}: {ex.Message}");
|
||||
}
|
||||
});
|
||||
}); // Parallel
|
||||
|
||||
lock (cacheLock)
|
||||
{
|
||||
latestDate = latestDateFromDb;
|
||||
|
||||
foreach (KeyValuePair<string, PricesByDate> kvp in fullReloads)
|
||||
{
|
||||
PricesByDate existing;
|
||||
if (priceCache.TryGetValue(kvp.Key, out existing) &&
|
||||
existing.MaxDate == currentMaxDates[kvp.Key])
|
||||
if (priceCache.TryGetValue(kvp.Key, out existing) && existing.MaxDate == currentMaxDates[kvp.Key])
|
||||
{
|
||||
priceCache[kvp.Key] = kvp.Value;
|
||||
}
|
||||
}
|
||||
|
||||
foreach (KeyValuePair<string, Price> kvp in singleUpdates)
|
||||
{
|
||||
PricesByDate pricesByDate;
|
||||
if (priceCache.TryGetValue(kvp.Key, out pricesByDate) &&
|
||||
pricesByDate.MaxDate == currentMaxDates[kvp.Key])
|
||||
if (priceCache.TryGetValue(kvp.Key, out pricesByDate) && pricesByDate.MaxDate == currentMaxDates[kvp.Key])
|
||||
{
|
||||
if (pricesByDate.ContainsKey(kvp.Value.Date))
|
||||
{
|
||||
pricesByDate.Remove(kvp.Value.Date);
|
||||
}
|
||||
pricesByDate.Add(kvp.Value.Date, kvp.Value);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
MDTrace.WriteLine(LogLevel.DEBUG,
|
||||
$"[GLPriceCache:Refresh] Full reloads: {fullReloads.Count}, Single updates: {singleUpdates.Count}");
|
||||
} // lock(cacheLock)
|
||||
MDTrace.WriteLine(LogLevel.DEBUG,$"[GLPriceCache:Refresh] Full reloads: {fullReloads.Count}, Single updates: {singleUpdates.Count}");
|
||||
}
|
||||
finally
|
||||
{
|
||||
@@ -386,16 +522,25 @@ namespace MarketData.Cache
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// GetLatestDate
|
||||
/// </summary>
|
||||
/// <returns></returns>
|
||||
public DateTime GetLatestDate()
|
||||
{
|
||||
lock (cacheLock)
|
||||
{
|
||||
if (Utility.IsEpoch(latestDate))
|
||||
{
|
||||
latestDate = PricingDA.GetLatestDate();
|
||||
}
|
||||
return latestDate;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// RefreshLatestDate
|
||||
/// </summary>
|
||||
public void RefreshLatestDate()
|
||||
{
|
||||
lock (cacheLock)
|
||||
@@ -404,36 +549,58 @@ namespace MarketData.Cache
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// GetMinCacheDate
|
||||
/// </summary>
|
||||
/// <param name="symbol"></param>
|
||||
/// <returns></returns>
|
||||
public DateTime GetMinCacheDate(string symbol)
|
||||
{
|
||||
lock (cacheLock)
|
||||
{
|
||||
PricesByDate symbolPrices;
|
||||
if (!priceCache.TryGetValue(symbol, out symbolPrices) || symbolPrices.Count == 0)
|
||||
{
|
||||
return Utility.Epoch;
|
||||
}
|
||||
return symbolPrices.MinDate;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// GetPrices
|
||||
/// </summary>
|
||||
/// <param name="symbol"></param>
|
||||
/// <param name="endDate"></param>
|
||||
/// <param name="dayCount"></param>
|
||||
/// <returns></returns>
|
||||
public Prices GetPrices(string symbol, DateTime endDate, int dayCount)
|
||||
{
|
||||
lock (cacheLock)
|
||||
{
|
||||
PricesByDate pricesByDate;
|
||||
if (!priceCache.TryGetValue(symbol, out pricesByDate)) return new Prices();
|
||||
|
||||
DateGenerator dateGenerator = new DateGenerator();
|
||||
List<DateTime> historicalDates = dateGenerator.GenerateHistoricalDates(endDate, dayCount);
|
||||
|
||||
Prices result = new Prices();
|
||||
foreach (DateTime date in historicalDates)
|
||||
{
|
||||
if (pricesByDate.ContainsKey(date))
|
||||
{
|
||||
result.Add(pricesByDate[date]);
|
||||
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
} // lock(cacheLock)
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// GetPrice
|
||||
/// </summary>
|
||||
/// <param name="symbol"></param>
|
||||
/// <param name="date"></param>
|
||||
/// <returns></returns>
|
||||
public Price GetPrice(string symbol, DateTime date)
|
||||
{
|
||||
lock (cacheLock)
|
||||
@@ -445,6 +612,12 @@ namespace MarketData.Cache
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// ContainsPrice
|
||||
/// </summary>
|
||||
/// <param name="symbol"></param>
|
||||
/// <param name="date"></param>
|
||||
/// <returns></returns>
|
||||
public bool ContainsPrice(string symbol, DateTime date)
|
||||
{
|
||||
lock (cacheLock)
|
||||
@@ -453,17 +626,30 @@ namespace MarketData.Cache
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// ContainsPrice
|
||||
/// </summary>
|
||||
/// <param name="symbols"></param>
|
||||
/// <param name="date"></param>
|
||||
/// <returns></returns>
|
||||
public bool ContainsPrice(List<string> symbols, DateTime date)
|
||||
{
|
||||
if (symbols == null || symbols.Count == 0) return false;
|
||||
lock (cacheLock)
|
||||
{
|
||||
foreach (string symbol in symbols)
|
||||
{
|
||||
if (!ContainsPriceInternal(symbol, date)) return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// ContainsSymbol
|
||||
/// </summary>
|
||||
/// <param name="symbol"></param>
|
||||
/// <returns></returns>
|
||||
public bool ContainsSymbol(string symbol)
|
||||
{
|
||||
lock (cacheLock)
|
||||
@@ -472,7 +658,7 @@ namespace MarketData.Cache
|
||||
}
|
||||
}
|
||||
|
||||
// ── Private helpers ──────────────────────────────────────────────────────
|
||||
// -- Private helpers ------------------------------------------------------
|
||||
|
||||
// Must be called under cacheLock
|
||||
private void AddInternal(Price price)
|
||||
@@ -485,10 +671,17 @@ namespace MarketData.Cache
|
||||
priceCache[price.Symbol] = pricesByDate;
|
||||
}
|
||||
if (!pricesByDate.ContainsKey(price.Date))
|
||||
{
|
||||
pricesByDate.Add(price.Date, price);
|
||||
}
|
||||
}
|
||||
|
||||
// Must be called under cacheLock
|
||||
/// <summary>
|
||||
/// ContainsPriceInternal - mUST BE CALLED UNDER CACHELOCK
|
||||
/// </summary>
|
||||
/// <param name="symbol"></param>
|
||||
/// <param name="date"></param>
|
||||
/// <returns></returns>
|
||||
private bool ContainsPriceInternal(string symbol, DateTime date)
|
||||
{
|
||||
PricesByDate pricesByDate;
|
||||
@@ -496,12 +689,17 @@ namespace MarketData.Cache
|
||||
return pricesByDate.ContainsKey(date);
|
||||
}
|
||||
|
||||
// Must be called under cacheLock
|
||||
/// <summary>
|
||||
/// CountInternal - MUST BE CALLED UNDER CACHELOCK
|
||||
/// </summary>
|
||||
/// <returns></returns>
|
||||
private long CountInternal()
|
||||
{
|
||||
long count = 0;
|
||||
foreach (PricesByDate pricesByDate in priceCache.Values)
|
||||
{
|
||||
count += pricesByDate.Count;
|
||||
}
|
||||
return count;
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user