All,
I have been thinking about how to use Ignite.Net to support an affinity
co-located ingest pipeline that uses queue buffering to provide fault
tolerance and buffering for a flow of ingest packages.
At a high level, it looks like this:
Arrival pipeline: [Gateway] -> [PackageReceiver] -> [PackageCache, affinity
co-located with PackageProcessor]
Processing pipeline: [PackageCache] -> [PackageProcessor] ->
[ProcessedDataCache affinity co-located with PackageProcessor]
Essentially, I want a cache that look like this:
Public class CacheItem
{
Public DateTime date;
[AffinityKeyMapped]
public Guid AffinityKey;
public byte [] Package;
}
ICache<string, CacheTime> BufferQueue.
BufferQueue = ignite.GetOrCreateCache <string, CacheItem > (
new CacheConfiguration
{
Name = “BufferQueue”,
KeepBinaryInStore = true,
// Replicate the maps across nodes
CacheMode = CacheMode.Partitioned,
});
}
This queue will target a data region that is configured for persistency.
Inbound packages will arrive and be injected into the BufferQueue cache
from some client node context, like this:
public void HandleANewPackage(string key, Guid affinityKey, byte [] package)
{
BufferQueue.Put(key, new CacheItem() {data = DateTime.Now(), AffinityKey =
affinityKey, Package = package});
}
There will be a collection of server nodes that are responsible for the
cache.
This is all straightforward. The tricky bit is then processing the elements
in the BufferQueue cache.
The data is already on the server nodes, nicely co-located according to its
affinity. I want to have parallel processing logic that runs on the server
nodes that pulls elements from the buffer queue and processes them into
some other cache(s).
At this point I know I have a cache that may contain something needing to
be processed, but I don’t know their keys. I know it’s possible to have
logic running on each server node that does this (either as a Grid Service
or a Compute::Broadcast() lambda):
var cache = ignite.GetCache<string, CacheItem>("BufferQueue");
var cursor = cache.Query(new ScanQuery<string, CacheItem >(new QueryFilter
()));
foreach (var cacheEntry in cursor)
ProcessItem(CacheEntry);
…but I am not sure how to restrict the elements in the cache returned to
the query to be only those entries affinity co-located with the server
asking for them.
Is this so obvious that it just works and does not need documentation, or
is this not possible and I should run the processing context from a client
node context (as above) and pay the penalty of extracting the packages from
the cache with cache.Query() and then resubmitting them using an affinity
aware method like AffinityRun()?
Thanks,
Raymond.
