I have a couple of processes that poll different data sources for some specific kind of information. They poll it quite often and do it in the background so when I need this information it is readily available and doesn't require a roundtrip that will waste time.
Sample code will look like this:
public class JournalBackgroundPoller
{
private readonly int _clusterSize;
private readonly IConfiguration _configuration;
Dictionary<int, string> _journalAddresses;
private readonly Random _localRandom;
private readonly Task _runHolder;
internal readonly ConcurrentDictionary<int, List<JournalEntryResponseItem>> ResultsBuffer = new ConcurrentDictionary<int, List<JournalEntryResponseItem>>();
public JournalBackgroundPoller(IConfiguration configuration)
{
_localRandom = new Random();
_configuration = configuration;
_clusterSize = 20;//for the sake of demo
_journalAddresses = //{{1, "SOME ADDR1"}, {2, "SOME ADDR 2"}};
_runHolder = BuildAndRun();
}
private Task BuildAndRun()
{
var pollingTasks = new List<Task>();
var buffer = new BroadcastBlock<JournalResponsesWrapper>(item => item);
PopulateShardsRegistry();
foreach (var js in _journalAddresses)
{
var dataProcessor = new TransformBlock<JournalResponsesWrapper, JournalResponsesWrapper>(NormalizeValues,
new ExecutionDataflowBlockOptions
{ MaxDegreeOfParallelism = 1, EnsureOrdered = true, BoundedCapacity = 1 });
var dataStorer = new ActionBlock<JournalResponsesWrapper>(StoreValuesInBuffer,
new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 1, EnsureOrdered = true, BoundedCapacity = 2 });
buffer.LinkTo(dataProcessor, wrapper => wrapper.JournalDataSource.Key == js.Key);
dataProcessor.LinkTo(dataStorer);
dataProcessor.LinkTo(DataflowBlock.NullTarget<JournalResponsesWrapper>());
pollingTasks.Add(PollInfinitely(js, buffer));
}
var r = Task.WhenAll(pollingTasks);
return r;
}
private void PopulateShardsRegistry()
{
try
{
for (int i = 0; i < _clusterSize; i++)
{
var _ = ResultsBuffer.GetOrAdd(i, ix => new List<JournalEntryResponseItem>());
}
}
catch (Exception e)
{
Console.WriteLine("Could `t initialize shards registry");
}
}
private async Task PollInfinitely(KeyValuePair<int, string> dataSourceInfo, BroadcastBlock<JournalResponsesWrapper> buffer)
{
while (true)
{
try
{
//here we create a client and get a big list of journal entries, ~200k from one source. below is dummy code
var journalEntries = new List<JournalEntryResponseItem>(200000);
buffer.Post(
new JournalResponsesWrapper { JournalDataSource = dataSourceInfo, JournalEntryResponseItems = journalEntries });
}
catch (Exception ex)
{
Console.WriteLine($"Polling {dataSourceInfo.Value} threw an exception, overwriting with empty data");
buffer.Post(
new JournalResponsesWrapper { JournalDataSource = dataSourceInfo, JournalEntryResponseItems = new List<JournalEntryResponseItem>() });
}
await Task.Delay(_localRandom.Next(400, 601));
}
}
private JournalResponsesWrapper NormalizeValues(JournalResponsesWrapper input)
{
try
{
if (input.JournalEntryResponseItems == null || !input.JournalEntryResponseItems.Any())
{
return input;
}
foreach (var journalEntry in input.JournalEntryResponseItems)
{
//do some transformations here
}
return input;
}
catch (Exception ex)
{
Console.WriteLine($"Normalization failed for cluster {input.JournalDataSource.Value}, please review!");
return null;
}
}
private void StoreValuesInBuffer(JournalResponsesWrapper input)
{
try
{
ResultsBuffer[input.JournalDataSource.Key] = input.JournalEntryResponseItems;
}
catch (Exception ex)
{
Console.WriteLine($"Could not write content to dictionary");
}
}
}
For simplicity journal related entities will look like this:
class JournalEntryResponseItem
{
public string SomeProperty1 { get; set; }
public string SomeProperty2 { get; set; }
}
class JournalResponsesWrapper
{
public KeyValuePair<int, string> JournalDataSource { get; set; }
public List<JournalEntryResponseItem> JournalEntryResponseItems { get; set; }
}
The global problem with the code provided is obviously that I'm creating a relatively big amount of objects that might end up in LOH in short period of time. Data sources always provide up to date entries so i don't need to keep the older ones (nor that i can do it as they are not distinguished). My question is whether it is possible to optimize the memory usage, object creation and replacement roundtrips so I can reduce the frequency of garbage collection? Right now by the looks of it garbage collection happens every ~5-10 seconds.
UPD 1: I access data via ResultsBuffer and can read the same set multiple times before it's refreshed. It's not guaranteed that one particular data set will be read only once (or read at all). My big objects are List<JournalEntryResponseItem> instances, initially coming from datasource and then saved to ResultsBuffer .
UPD 2: Data sources have only one endpoint that returns all entities in this "shard" at once, I can't apply filtering during request. Response entities do not have unique keys/identifiers.
UPD 3: Some answers suggest to measure/profile the app first. While this is perfectly valid suggestion in this particular case it's clearly memory/GC related because of the following observations:
- Visual throttling happens exactly at the moment when apps RAM consumption goes down sharply after growing steadily for some time.
- If I add X more journal sources apps' memory will grow until it takes all free memory on the server and then there is an even longer freeze (1-3 seconds) after which memory goes down sharply and app keeps working until it hits memory limit again.
