GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Make SoA cache-managed document storage

Browse source
This commit is contained in:
Meta Vens 2026-06-13 01:44:05 +01:00
commit 5428870544
10 changed files with 599 additions and 787 deletions
Download patch file Download diff file Expand all files Collapse all files

67
src/GameCult.Caching/CultCache.cs
View file

@ -1159,6 +1159,8 @@ namespace GameCult.Caching
private readonly ConcurrentDictionary<(Type Type, string Alias), ConcurrentDictionary<string, string>> _indexMaps = new();
private readonly ConcurrentDictionary<Type, string> _globalKeys = new();
private readonly ConditionalWeakTable<object, DocumentHandleBox> _documentHandles = new();
private readonly Subject<object> _changes = new();
private readonly CultCacheSoaStore _soa = new();
private ILogger _logger = new NullLogger();
private bool _hasUnflushedMutations;
@ -1222,6 +1224,46 @@ namespace GameCult.Caching
/// </summary>
public CultDocumentRegistry Registry => _registry;
/// <summary>
/// Opens a reactive POCO presentation for one cache-managed document.
/// </summary>
public CultManagedDocument<T> Document<T>(CultRecordKey key) where T : class
{
return new CultManagedDocument<T>(
key,
() => Get<T>(key),
async value => { await UpsertAsync(value, new CultRecordHandle<T>(key)).ConfigureAwait(false); },
WatchRecord<T>(key)
.Where(change => change.Document != null)
.Select(change => change.Document!));
}
/// <summary>
/// Watches all local cache changes assignable to the requested document type.
/// </summary>
public Observable<CultCacheDocumentChange<T>> Watch<T>() where T : class
{
return _changes
.Where(change => change is CultCacheDocumentChange<T>)
.Select(change => (CultCacheDocumentChange<T>)change);
}
/// <summary>
/// Watches one local cache record.
/// </summary>
public Observable<CultCacheDocumentChange<T>> WatchRecord<T>(CultRecordKey key) where T : class
{
return Watch<T>().Where(change => change.Key.Equals(key));
}
/// <summary>
/// Gets the cache-owned structure-of-arrays table for a document type.
/// </summary>
public CultSoaTable<T> Soa<T>() where T : class
{
return _soa.Snapshot<T>();
}
/// <summary>
/// Attaches a backing store to this cache.
/// </summary>
@ -1522,6 +1564,8 @@ namespace GameCult.Caching
{
store.Dispose();
}
_changes.Dispose();
}
internal CultStoredDocument CreateStoredDocument(Type documentType, object document, CultRecordKey? key = null, string? storedAt = null)
@ -1550,6 +1594,7 @@ namespace GameCult.Caching
_entries[stored.Key.Value] = stored;
_documentHandles.Remove(stored.Document);
_documentHandles.Add(stored.Document, new DocumentHandleBox(stored.Key));
_soa.Upsert(stored);
AddIndexes(stored);
foreach (var store in _backingStores)
@ -1569,6 +1614,8 @@ namespace GameCult.Caching
RecomputeDirtyState();
}
PublishChange(stored, existing?.Document);
if (raiseUpdate)
{
OnUpdate?.Invoke(existing?.Document, stored.Document);
@ -1590,6 +1637,7 @@ namespace GameCult.Caching
RemoveIndexes(existing);
_documentHandles.Remove(existing.Document);
_soa.Remove(existing);
foreach (var store in _backingStores)
{
@ -1608,6 +1656,8 @@ namespace GameCult.Caching
RecomputeDirtyState();
}
PublishChange(existing, existing.Document, removed: true);
if (raiseUpdate)
{
OnUpdate?.Invoke(existing.Document, null);
@ -1729,6 +1779,23 @@ namespace GameCult.Caching
? _hasUnflushedMutations
: _backingStores.Any(store => store.IsDirty);
}
private void PublishChange(CultStoredDocument stored, object? previousDocument, bool removed = false)
{
var changeType = typeof(CultCacheDocumentChange<>).MakeGenericType(stored.Descriptor.DocumentType);
var change = Activator.CreateInstance(
changeType,
removed ? CultCacheDocumentChangeKind.Removed :
previousDocument == null ? CultCacheDocumentChangeKind.Added :
CultCacheDocumentChangeKind.Updated,
stored.Key,
removed ? null : stored.Document,
previousDocument);
if (change != null)
{
_changes.OnNext(change);
}
}
}
/// <summary>

415
src/GameCult.Caching/CultManagedDocument.cs Normal file
View file

@ -0,0 +1,415 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using System.Threading.Tasks;
using R3;
namespace GameCult.Caching
{
/// <summary>
/// Describes how a CultCache-managed document changed.
/// </summary>
public enum CultCacheDocumentChangeKind
{
/// <summary>
/// A document was added.
/// </summary>
Added,
/// <summary>
/// A document was updated.
/// </summary>
Updated,
/// <summary>
/// A document was removed.
/// </summary>
Removed
}
/// <summary>
/// One reactive CultCache document change.
/// </summary>
public sealed class CultCacheDocumentChange<T> where T : class
{
/// <summary>
/// Creates a document change.
/// </summary>
public CultCacheDocumentChange(
CultCacheDocumentChangeKind kind,
CultRecordKey key,
T? document,
T? previousDocument)
{
Kind = kind;
Key = key;
Document = document;
PreviousDocument = previousDocument;
}
/// <summary>
/// Gets the change kind.
/// </summary>
public CultCacheDocumentChangeKind Kind { get; }
/// <summary>
/// Gets the document record key.
/// </summary>
public CultRecordKey Key { get; }
/// <summary>
/// Gets the current POCO presentation, when present.
/// </summary>
public T? Document { get; }
/// <summary>
/// Gets the previous POCO presentation, when present.
/// </summary>
public T? PreviousDocument { get; }
}
/// <summary>
/// Reactive POCO presentation for one CultCache-managed document.
/// </summary>
public sealed class CultManagedDocument<T> where T : class
{
private readonly Func<T?> _read;
private readonly Func<T, Task> _commit;
private readonly Observable<T> _watch;
/// <summary>
/// Creates a managed document presentation.
/// </summary>
public CultManagedDocument(
CultRecordKey key,
Func<T?> read,
Func<T, Task> commit,
Observable<T> watch)
{
Key = key;
_read = read ?? throw new ArgumentNullException(nameof(read));
_commit = commit ?? throw new ArgumentNullException(nameof(commit));
_watch = watch ?? throw new ArgumentNullException(nameof(watch));
}
/// <summary>
/// Gets the document record key.
/// </summary>
public CultRecordKey Key { get; }
/// <summary>
/// Gets the current POCO presentation.
/// </summary>
public T? Value => _read();
/// <summary>
/// Watches committed values for this document.
/// </summary>
public Observable<T> Watch()
{
return _watch;
}
/// <summary>
/// Commits the current POCO value through the owner-supplied commit path.
/// </summary>
public Task CommitAsync()
{
var value = Value ?? throw new InvalidOperationException($"Document '{Key.Value}' is not present.");
return _commit(value);
}
/// <summary>
/// Replaces the current value through the owner-supplied commit path.
/// </summary>
public Task ReplaceAsync(T value)
{
if (value == null) throw new ArgumentNullException(nameof(value));
return _commit(value);
}
}
/// <summary>
/// CPU-local structure-of-arrays table owned by CultCache for one document type.
/// </summary>
public sealed class CultSoaTable<TDocument> where TDocument : class
{
private readonly Dictionary<string, object> _columns;
internal CultSoaTable(
CultRecordKey[] keys,
TDocument[] documents,
Dictionary<string, object> columns)
{
Keys = keys;
Documents = documents;
_columns = columns;
}
/// <summary>
/// Gets record keys in row order.
/// </summary>
public IReadOnlyList<CultRecordKey> Keys { get; }
/// <summary>
/// Gets POCO presentations in row order.
/// </summary>
public IReadOnlyList<TDocument> Documents { get; }
/// <summary>
/// Gets the row count.
/// </summary>
public int Count => Keys.Count;
/// <summary>
/// Gets a contiguous unmanaged field/property column.
/// </summary>
public CultSoaColumn<TValue> Column<TValue>(string name) where TValue : unmanaged
{
if (!_columns.TryGetValue(name, out var column))
{
throw new KeyNotFoundException($"SoA table for {typeof(TDocument).Name} has no '{name}' column.");
}
if (column is not TValue[] values)
{
throw new InvalidOperationException(
$"SoA table column '{name}' stores {column.GetType().GetElementType()?.FullName}, not {typeof(TValue).FullName}.");
}
return new CultSoaColumn<TValue>(name, values);
}
}
/// <summary>
/// Contiguous typed SoA column.
/// </summary>
public readonly struct CultSoaColumn<TValue> where TValue : unmanaged
{
internal CultSoaColumn(string name, TValue[] values)
{
Name = name;
Values = values;
}
/// <summary>
/// Gets the column name.
/// </summary>
public string Name { get; }
/// <summary>
/// Gets the backing array.
/// </summary>
public TValue[] Values { get; }
/// <summary>
/// Gets the backing array as a span.
/// </summary>
public Span<TValue> Span => Values.AsSpan();
}
internal sealed class CultCacheSoaStore
{
private readonly Dictionary<Type, CultCacheSoaTypeTable> _tables = new();
public void Upsert(CultStoredDocument stored)
{
Table(stored.Descriptor.DocumentType).Upsert(stored.Key, stored.Document);
}
public void Remove(CultStoredDocument stored)
{
if (_tables.TryGetValue(stored.Descriptor.DocumentType, out var table))
{
table.Remove(stored.Key);
}
}
public CultSoaTable<TDocument> Snapshot<TDocument>() where TDocument : class
{
return Table(typeof(TDocument)).Snapshot<TDocument>();
}
private CultCacheSoaTypeTable Table(Type documentType)
{
if (!_tables.TryGetValue(documentType, out var table))
{
table = new CultCacheSoaTypeTable(documentType);
_tables[documentType] = table;
}
return table;
}
}
internal sealed class CultCacheSoaTypeTable
{
private readonly Type _documentType;
private readonly List<CultRecordKey> _keys = new();
private readonly List<object> _documents = new();
private readonly Dictionary<string, CultCacheSoaMember> _members;
private readonly Dictionary<string, object> _columns;
private readonly Dictionary<string, int> _rowsByKey = new(StringComparer.Ordinal);
public CultCacheSoaTypeTable(Type documentType)
{
_documentType = documentType;
_members = CultCacheSoaMember.Discover(documentType)
.ToDictionary(member => member.Name, StringComparer.Ordinal);
_columns = _members.Values.ToDictionary(
member => member.Name,
member => Array.CreateInstance(member.ValueType, 0) as object,
StringComparer.Ordinal);
}
public void Upsert(CultRecordKey key, object document)
{
if (_rowsByKey.TryGetValue(key.Value, out var row))
{
_documents[row] = document;
}
else
{
row = _keys.Count;
_rowsByKey[key.Value] = row;
_keys.Add(key);
_documents.Add(document);
EnsureColumnCapacity(_keys.Count);
}
WriteRow(row, document);
}
public void Remove(CultRecordKey key)
{
if (!_rowsByKey.TryGetValue(key.Value, out var row))
{
return;
}
var lastRow = _keys.Count - 1;
if (row != lastRow)
{
_keys[row] = _keys[lastRow];
_documents[row] = _documents[lastRow];
_rowsByKey[_keys[row].Value] = row;
CopyRow(lastRow, row);
}
_rowsByKey.Remove(key.Value);
_keys.RemoveAt(lastRow);
_documents.RemoveAt(lastRow);
}
public CultSoaTable<TDocument> Snapshot<TDocument>() where TDocument : class
{
if (!typeof(TDocument).IsAssignableFrom(_documentType))
{
throw new InvalidOperationException($"{_documentType.FullName} is not assignable to {typeof(TDocument).FullName}.");
}
var count = _keys.Count;
var columns = new Dictionary<string, object>(StringComparer.Ordinal);
foreach (var pair in _columns)
{
var member = _members[pair.Key];
var source = (Array)pair.Value;
var target = Array.CreateInstance(member.ValueType, count);
Array.Copy(source, target, count);
columns[pair.Key] = target;
}
return new CultSoaTable<TDocument>(
_keys.ToArray(),
_documents.Cast<TDocument>().ToArray(),
columns);
}
private void EnsureColumnCapacity(int count)
{
foreach (var member in _members.Values)
{
var column = (Array)_columns[member.Name];
if (column.Length >= count)
{
continue;
}
var nextLength = Math.Max(count, Math.Max(4, column.Length * 2));
var next = Array.CreateInstance(member.ValueType, nextLength);
Array.Copy(column, next, column.Length);
_columns[member.Name] = next;
}
}
private void WriteRow(int row, object document)
{
foreach (var member in _members.Values)
{
((Array)_columns[member.Name]).SetValue(member.Get(document), row);
}
}
private void CopyRow(int sourceRow, int targetRow)
{
foreach (var member in _members.Values)
{
var column = (Array)_columns[member.Name];
column.SetValue(column.GetValue(sourceRow), targetRow);
}
}
}
internal sealed class CultCacheSoaMember
{
private CultCacheSoaMember(string name, Type valueType, Func<object, object?> getter)
{
Name = name;
ValueType = valueType;
Get = getter;
}
public string Name { get; }
public Type ValueType { get; }
public Func<object, object?> Get { get; }
public static IEnumerable<CultCacheSoaMember> Discover(Type documentType)
{
const BindingFlags flags = BindingFlags.Instance | BindingFlags.Public;
foreach (var field in documentType.GetFields(flags).OrderBy(field => field.MetadataToken))
{
if (IsColumnType(field.FieldType))
{
yield return new CultCacheSoaMember(field.Name, field.FieldType, field.GetValue);
}
}
foreach (var property in documentType.GetProperties(flags).OrderBy(property => property.MetadataToken))
{
if (property.GetMethod != null &&
property.GetIndexParameters().Length == 0 &&
IsColumnType(property.PropertyType))
{
yield return new CultCacheSoaMember(property.Name, property.PropertyType, property.GetValue);
}
}
}
private static bool IsColumnType(Type type)
{
return type.IsEnum ||
type == typeof(byte) ||
type == typeof(sbyte) ||
type == typeof(short) ||
type == typeof(ushort) ||
type == typeof(int) ||
type == typeof(uint) ||
type == typeof(long) ||
type == typeof(ulong) ||
type == typeof(float) ||
type == typeof(double) ||
type == typeof(bool);
}
}
}

630
src/GameCult.Caching/CultSoa.cs
View file

@ -1,630 +0,0 @@
using System;
using System.Collections.Generic;
using System.IO;
using System.Reflection;
using System.Linq;
using System.Runtime.InteropServices;
using System.Text;
using System.Threading.Tasks;
[assembly: GameCult.Caching.CultGeneratedDocumentMetadataProvider(typeof(GameCult.Caching.CultSoaGeneratedDocumentMetadataProvider))]
namespace GameCult.Caching
{
/// <summary>
/// CultCache document that stores a CPU-local structure-of-arrays chunk.
/// </summary>
[CultDocument("gamecult.soa_chunk", "gamecult.soa_chunk.v1")]
public sealed class CultSoaChunkDocument
{
/// <summary>
/// Gets or sets the stable chunk identifier.
/// </summary>
[CultName]
public string ChunkId { get; set; } = string.Empty;
/// <summary>
/// Gets or sets the layout/archetype identifier shared by chunks with the same columns.
/// </summary>
[CultIndex("archetype")]
public string ArchetypeId { get; set; } = string.Empty;
/// <summary>
/// Gets or sets the number of live rows in this chunk.
/// </summary>
public int Count { get; set; }
/// <summary>
/// Gets or sets the live row entity ids. The array length is the chunk capacity.
/// </summary>
public ulong[] EntityIds { get; set; } = Array.Empty<ulong>();
/// <summary>
/// Gets or sets the contiguous component columns owned by this chunk.
/// </summary>
public CultSoaColumnDocument[] Columns { get; set; } = Array.Empty<CultSoaColumnDocument>();
}
/// <summary>
/// One contiguous column inside a structure-of-arrays chunk.
/// </summary>
public sealed class CultSoaColumnDocument
{
/// <summary>
/// Gets or sets the stable column name, usually a component field path.
/// </summary>
public string Name { get; set; } = string.Empty;
/// <summary>
/// Gets or sets the CLR element type name stored in the column.
/// </summary>
public string ElementType { get; set; } = string.Empty;
/// <summary>
/// Gets or sets the size of one element in bytes.
/// </summary>
public int ElementByteLength { get; set; }
/// <summary>
/// Gets or sets the contiguous column payload.
/// </summary>
public byte[] Data { get; set; } = Array.Empty<byte>();
}
/// <summary>
/// Runtime view over a CultCache SoA chunk document.
/// </summary>
public sealed class CultSoaChunk
{
private readonly Dictionary<string, CultSoaColumnDocument> _columns;
private CultSoaChunk(CultSoaChunkDocument document)
{
Document = document ?? throw new ArgumentNullException(nameof(document));
ValidateDocument(document);
_columns = document.Columns.ToDictionary(column => column.Name, StringComparer.Ordinal);
}
/// <summary>
/// Gets the underlying CultCache document.
/// </summary>
public CultSoaChunkDocument Document { get; }
/// <summary>
/// Gets the stable chunk identifier.
/// </summary>
public string ChunkId => Document.ChunkId;
/// <summary>
/// Gets the layout/archetype identifier.
/// </summary>
public string ArchetypeId => Document.ArchetypeId;
/// <summary>
/// Gets the number of live rows.
/// </summary>
public int Count => Document.Count;
/// <summary>
/// Gets the maximum row count this chunk can hold without reallocating.
/// </summary>
public int Capacity => Document.EntityIds.Length;
/// <summary>
/// Gets all entity ids, including unused capacity.
/// </summary>
public Span<ulong> EntityIds => Document.EntityIds.AsSpan();
/// <summary>
/// Gets live entity ids.
/// </summary>
public Span<ulong> ActiveEntityIds => Document.EntityIds.AsSpan(0, Document.Count);
/// <summary>
/// Creates an empty SoA chunk with the supplied capacity.
/// </summary>
public static CultSoaChunk Create(string chunkId, string archetypeId, int capacity)
{
if (string.IsNullOrWhiteSpace(chunkId)) throw new ArgumentException("Value must be non-empty.", nameof(chunkId));
if (string.IsNullOrWhiteSpace(archetypeId)) throw new ArgumentException("Value must be non-empty.", nameof(archetypeId));
if (capacity < 0) throw new ArgumentOutOfRangeException(nameof(capacity));
return new CultSoaChunk(new CultSoaChunkDocument
{
ChunkId = chunkId,
ArchetypeId = archetypeId,
Count = 0,
EntityIds = new ulong[capacity],
Columns = Array.Empty<CultSoaColumnDocument>()
});
}
/// <summary>
/// Wraps a persisted SoA chunk document.
/// </summary>
public static CultSoaChunk Wrap(CultSoaChunkDocument document)
{
return new CultSoaChunk(document);
}
/// <summary>
/// Sets the live row count after validating all columns have enough capacity.
/// </summary>
public void SetCount(int count)
{
if (count < 0 || count > Capacity)
{
throw new ArgumentOutOfRangeException(nameof(count));
}
Document.Count = count;
}
/// <summary>
/// Appends one entity row and returns its row index.
/// </summary>
public int AddEntity(ulong entityId)
{
if (Document.Count >= Capacity)
{
throw new InvalidOperationException($"SoA chunk '{ChunkId}' is full.");
}
var row = Document.Count;
Document.EntityIds[row] = entityId;
Document.Count++;
return row;
}
/// <summary>
/// Adds a contiguous typed column.
/// </summary>
public CultSoaColumn<T> AddColumn<T>(string name) where T : unmanaged
{
if (string.IsNullOrWhiteSpace(name)) throw new ArgumentException("Value must be non-empty.", nameof(name));
if (_columns.ContainsKey(name))
{
throw new InvalidOperationException($"SoA chunk '{ChunkId}' already has a '{name}' column.");
}
var column = new CultSoaColumnDocument
{
Name = name,
ElementType = CultSoaTypeName.For<T>(),
ElementByteLength = Marshal.SizeOf<T>(),
Data = new byte[checked(Capacity * Marshal.SizeOf<T>())]
};
var columns = Document.Columns.ToList();
columns.Add(column);
Document.Columns = columns.ToArray();
_columns[name] = column;
return new CultSoaColumn<T>(this, column);
}
/// <summary>
/// Gets an existing typed column.
/// </summary>
public CultSoaColumn<T> Column<T>(string name) where T : unmanaged
{
if (!_columns.TryGetValue(name, out var column))
{
throw new KeyNotFoundException($"SoA chunk '{ChunkId}' has no '{name}' column.");
}
var expectedType = CultSoaTypeName.For<T>();
var expectedByteLength = Marshal.SizeOf<T>();
if (!string.Equals(column.ElementType, expectedType, StringComparison.Ordinal) ||
column.ElementByteLength != expectedByteLength)
{
throw new InvalidOperationException(
$"SoA column '{name}' stores {column.ElementType}/{column.ElementByteLength} bytes, not {expectedType}/{expectedByteLength}.");
}
var expectedLength = checked(Capacity * expectedByteLength);
if (column.Data.Length != expectedLength)
{
throw new InvalidOperationException(
$"SoA column '{name}' has {column.Data.Length} bytes, expected {expectedLength} for chunk capacity {Capacity}.");
}
return new CultSoaColumn<T>(this, column);
}
private static void ValidateDocument(CultSoaChunkDocument document)
{
if (string.IsNullOrWhiteSpace(document.ChunkId))
{
throw new ArgumentException("SoA chunk documents require a chunk id.", nameof(document));
}
if (string.IsNullOrWhiteSpace(document.ArchetypeId))
{
throw new ArgumentException("SoA chunk documents require an archetype id.", nameof(document));
}
document.EntityIds ??= Array.Empty<ulong>();
document.Columns ??= Array.Empty<CultSoaColumnDocument>();
if (document.Count < 0 || document.Count > document.EntityIds.Length)
{
throw new ArgumentOutOfRangeException(nameof(document), "SoA chunk count must fit inside entity id capacity.");
}
foreach (var column in document.Columns)
{
if (column == null) throw new ArgumentException("SoA chunk columns cannot contain null entries.", nameof(document));
if (string.IsNullOrWhiteSpace(column.Name)) throw new ArgumentException("SoA columns require names.", nameof(document));
if (string.IsNullOrWhiteSpace(column.ElementType)) throw new ArgumentException($"SoA column '{column.Name}' requires an element type.", nameof(document));
if (column.ElementByteLength <= 0) throw new ArgumentOutOfRangeException(nameof(document), $"SoA column '{column.Name}' requires a positive element byte length.");
column.Data ??= Array.Empty<byte>();
var expectedLength = checked(document.EntityIds.Length * column.ElementByteLength);
if (column.Data.Length != expectedLength)
{
throw new ArgumentException(
$"SoA column '{column.Name}' has {column.Data.Length} bytes, expected {expectedLength}.",
nameof(document));
}
}
}
}
/// <summary>
/// Typed contiguous view over one SoA column.
/// </summary>
public readonly struct CultSoaColumn<T> where T : unmanaged
{
private readonly CultSoaChunk _owner;
private readonly CultSoaColumnDocument _column;
internal CultSoaColumn(CultSoaChunk owner, CultSoaColumnDocument column)
{
_owner = owner;
_column = column;
}
/// <summary>
/// Gets the column name.
/// </summary>
public string Name => _column.Name;
/// <summary>
/// Gets the full column capacity as a typed span.
/// </summary>
public Span<T> Span => MemoryMarshal.Cast<byte, T>(_column.Data.AsSpan());
/// <summary>
/// Gets the live rows as a typed span.
/// </summary>
public Span<T> ActiveSpan => Span.Slice(0, _owner.Count);
}
/// <summary>
/// Convenience extensions for storing SoA chunks in CultCache.
/// </summary>
public static class CultSoaCacheExtensions
{
/// <summary>
/// Adds or replaces a SoA chunk document in a CultCache.
/// </summary>
public static Task<CultRecordHandle<CultSoaChunkDocument>> UpsertSoaChunkAsync(
this CultCache cache,
CultSoaChunk chunk,
CultRecordHandle<CultSoaChunkDocument>? handle = null)
{
if (cache == null) throw new ArgumentNullException(nameof(cache));
if (chunk == null) throw new ArgumentNullException(nameof(chunk));
return cache.UpsertAsync(chunk.Document, handle);
}
/// <summary>
/// Gets a SoA chunk view by record key.
/// </summary>
public static CultSoaChunk? GetSoaChunk(this CultCache cache, CultRecordKey key)
{
if (cache == null) throw new ArgumentNullException(nameof(cache));
var document = cache.Get<CultSoaChunkDocument>(key);
return document == null ? null : CultSoaChunk.Wrap(document);
}
/// <summary>
/// Projects ordinary typed CultCache documents into contiguous unmanaged columns for hot CPU loops.
/// </summary>
public static CultSoaDocumentTable<TDocument> ProjectSoa<TDocument>(this CultCache cache) where TDocument : class
{
if (cache == null) throw new ArgumentNullException(nameof(cache));
return CultSoaDocumentTable<TDocument>.From(cache);
}
}
/// <summary>
/// Structure-of-arrays projection over ordinary CultCache documents.
/// </summary>
public sealed class CultSoaDocumentTable<TDocument> where TDocument : class
{
private readonly Dictionary<string, object> _columns;
private CultSoaDocumentTable(
CultRecordKey[] keys,
TDocument[] documents,
Dictionary<string, object> columns)
{
Keys = keys;
Documents = documents;
_columns = columns;
}
/// <summary>
/// Gets the projected record keys in row order.
/// </summary>
public IReadOnlyList<CultRecordKey> Keys { get; }
/// <summary>
/// Gets the projected document references in row order.
/// </summary>
public IReadOnlyList<TDocument> Documents { get; }
/// <summary>
/// Gets the row count.
/// </summary>
public int Count => Documents.Count;
/// <summary>
/// Builds a contiguous projection from all documents of this type in the cache.
/// </summary>
public static CultSoaDocumentTable<TDocument> From(CultCache cache)
{
if (cache == null) throw new ArgumentNullException(nameof(cache));
var stored = cache.AllStoredDocuments
.Where(entry => entry.Document is TDocument)
.OrderBy(entry => entry.Key.Value, StringComparer.Ordinal)
.ToArray();
var keys = stored.Select(entry => entry.Key).ToArray();
var documents = stored.Select(entry => (TDocument)entry.Document).ToArray();
var columns = new Dictionary<string, object>(StringComparer.Ordinal);
foreach (var member in CultSoaDocumentMember.Discover(typeof(TDocument)))
{
columns[member.Name] = member.Project(documents);
}
return new CultSoaDocumentTable<TDocument>(keys, documents, columns);
}
/// <summary>
/// Gets a contiguous projected column by field or property name.
/// </summary>
public CultSoaDocumentColumn<TValue> Column<TValue>(string name) where TValue : unmanaged
{
if (!_columns.TryGetValue(name, out var column))
{
throw new KeyNotFoundException($"SoA projection for {typeof(TDocument).Name} has no '{name}' column.");
}
if (column is not TValue[] values)
{
throw new InvalidOperationException(
$"SoA projection column '{name}' stores {column.GetType().GetElementType()?.FullName}, not {typeof(TValue).FullName}.");
}
return new CultSoaDocumentColumn<TValue>(name, values);
}
}
/// <summary>
/// Contiguous projected column from ordinary CultCache documents.
/// </summary>
public readonly struct CultSoaDocumentColumn<TValue> where TValue : unmanaged
{
internal CultSoaDocumentColumn(string name, TValue[] values)
{
Name = name;
Values = values;
}
/// <summary>
/// Gets the projected field or property name.
/// </summary>
public string Name { get; }
/// <summary>
/// Gets the backing array.
/// </summary>
public TValue[] Values { get; }
/// <summary>
/// Gets the column as a contiguous span.
/// </summary>
public Span<TValue> Span => Values.AsSpan();
}
internal sealed class CultSoaDocumentMember
{
private CultSoaDocumentMember(string name, Type valueType, Func<object, object?> getter)
{
Name = name;
ValueType = valueType;
Getter = getter;
}
public string Name { get; }
public Type ValueType { get; }
private Func<object, object?> Getter { get; }
public static IEnumerable<CultSoaDocumentMember> Discover(Type documentType)
{
const BindingFlags flags = BindingFlags.Instance | BindingFlags.Public;
foreach (var field in documentType.GetFields(flags).OrderBy(field => field.MetadataToken))
{
if (IsSupportedColumnType(field.FieldType))
{
yield return new CultSoaDocumentMember(field.Name, field.FieldType, field.GetValue);
}
}
foreach (var property in documentType.GetProperties(flags).OrderBy(property => property.MetadataToken))
{
if (property.GetMethod != null &&
property.GetIndexParameters().Length == 0 &&
IsSupportedColumnType(property.PropertyType))
{
yield return new CultSoaDocumentMember(property.Name, property.PropertyType, property.GetValue);
}
}
}
public object Project(IReadOnlyList<object> documents)
{
var array = Array.CreateInstance(ValueType, documents.Count);
for (var index = 0; index < documents.Count; index++)
{
array.SetValue(Getter(documents[index]) ?? Activator.CreateInstance(ValueType), index);
}
return array;
}
private static bool IsSupportedColumnType(Type type)
{
return type.IsEnum ||
type == typeof(byte) ||
type == typeof(sbyte) ||
type == typeof(short) ||
type == typeof(ushort) ||
type == typeof(int) ||
type == typeof(uint) ||
type == typeof(long) ||
type == typeof(ulong) ||
type == typeof(float) ||
type == typeof(double) ||
type == typeof(bool);
}
}
internal static class CultSoaTypeName
{
public static string For<T>() where T : unmanaged
{
var type = typeof(T);
if (type == typeof(byte)) return "u8";
if (type == typeof(sbyte)) return "i8";
if (type == typeof(short)) return "i16";
if (type == typeof(ushort)) return "u16";
if (type == typeof(int)) return "i32";
if (type == typeof(uint)) return "u32";
if (type == typeof(long)) return "i64";
if (type == typeof(ulong)) return "u64";
if (type == typeof(float)) return "f32";
if (type == typeof(double)) return "f64";
return type.AssemblyQualifiedName ?? type.FullName ?? type.Name;
}
}
/// <summary>
/// Built-in metadata and payload codec for CultCache SoA documents.
/// </summary>
public sealed class CultSoaGeneratedDocumentMetadataProvider : ICultGeneratedDocumentMetadataProvider
{
/// <inheritdoc />
public IEnumerable<CultGeneratedDocumentDefinition> GetDocumentDefinitions()
{
yield return new CultGeneratedDocumentDefinition(
typeof(CultSoaChunkDocument),
"gamecult.soa_chunk",
"gamecult.soa_chunk.v1",
isGlobal: false,
nameMember: nameof(CultSoaChunkDocument.ChunkId),
nameAccessor: document => ((CultSoaChunkDocument)document).ChunkId,
serializePayload: SerializeChunk,
deserializePayload: DeserializeChunk,
indexAccessors: new[]
{
new CultGeneratedDocumentIndexAccessor(
"archetype",
document => ((CultSoaChunkDocument)document).ArchetypeId)
},
members: new[]
{
new CultGeneratedDocumentMemberDefinition(nameof(CultSoaChunkDocument.ChunkId), 0, typeof(string).FullName!, false, false, null, true, null),
new CultGeneratedDocumentMemberDefinition(nameof(CultSoaChunkDocument.ArchetypeId), 1, typeof(string).FullName!, false, false, null, false, "archetype"),
new CultGeneratedDocumentMemberDefinition(nameof(CultSoaChunkDocument.Count), 2, typeof(int).FullName!, false, false, null, false, null),
new CultGeneratedDocumentMemberDefinition(nameof(CultSoaChunkDocument.EntityIds), 3, typeof(ulong[]).FullName!, false, false, null, false, null),
new CultGeneratedDocumentMemberDefinition(nameof(CultSoaChunkDocument.Columns), 4, typeof(CultSoaColumnDocument[]).FullName!, false, false, null, false, null)
});
}
private static byte[] SerializeChunk(object value)
{
var chunk = (CultSoaChunkDocument)value;
using var stream = new MemoryStream();
using var writer = new BinaryWriter(stream, Encoding.UTF8);
writer.Write(1);
writer.Write(chunk.ChunkId ?? string.Empty);
writer.Write(chunk.ArchetypeId ?? string.Empty);
writer.Write(chunk.Count);
var entityIds = chunk.EntityIds ?? Array.Empty<ulong>();
writer.Write(entityIds.Length);
foreach (var entityId in entityIds)
{
writer.Write(entityId);
}
var columns = chunk.Columns ?? Array.Empty<CultSoaColumnDocument>();
writer.Write(columns.Length);
foreach (var column in columns)
{
writer.Write(column.Name ?? string.Empty);
writer.Write(column.ElementType ?? string.Empty);
writer.Write(column.ElementByteLength);
var data = column.Data ?? Array.Empty<byte>();
writer.Write(data.Length);
writer.Write(data);
}
writer.Flush();
return stream.ToArray();
}
private static object DeserializeChunk(byte[] payload)
{
using var stream = new MemoryStream(payload);
using var reader = new BinaryReader(stream, Encoding.UTF8);
var version = reader.ReadInt32();
if (version != 1)
{
throw new InvalidOperationException($"Unsupported SoA chunk payload version '{version}'.");
}
var chunk = new CultSoaChunkDocument
{
ChunkId = reader.ReadString(),
ArchetypeId = reader.ReadString(),
Count = reader.ReadInt32()
};
var entityCount = reader.ReadInt32();
chunk.EntityIds = new ulong[entityCount];
for (var index = 0; index < entityCount; index++)
{
chunk.EntityIds[index] = reader.ReadUInt64();
}
var columnCount = reader.ReadInt32();
chunk.Columns = new CultSoaColumnDocument[columnCount];
for (var index = 0; index < columnCount; index++)
{
var column = new CultSoaColumnDocument
{
Name = reader.ReadString(),
ElementType = reader.ReadString(),
ElementByteLength = reader.ReadInt32()
};
var dataLength = reader.ReadInt32();
column.Data = reader.ReadBytes(dataLength);
chunk.Columns[index] = column;
}
return chunk;
}
}
}

48
src/GameCult.Caching/README.md
View file

@ -143,10 +143,11 @@ If you want the explicit assembly rite, the lower-level `CultCache` +
`SingleFileMessagePackBackingStore` path is still there. The helper is just the
blessed front door.
## CPU-Local SoA Access
## Managed Documents And SoA Storage
Gameplay code can keep authoring normal typed documents and let CultCache build
hot structure-of-arrays projections when a loop wants contiguous columns.
Gameplay code keeps authoring normal typed documents. CultCache owns the record
identity, reactive presentation, and CPU-local structure-of-arrays table behind
that POCO surface.
```csharp
[CultDocument("game.player_transform", "game.player_transform.v1")]
@ -158,7 +159,22 @@ public sealed class PlayerTransform
public int Health;
}
var table = cache.ProjectSoa<PlayerTransform>();
var player = cache.Document<PlayerTransform>(new CultRecordKey("player:alice"));
await player.ReplaceAsync(new PlayerTransform
{
Name = "Alice",
PositionX = 4,
PositionY = 8,
Health = 100
});
using var subscription = player.Watch().Subscribe(current => Render(current));
player.Value!.Health = 90;
await player.CommitAsync();
var table = cache.Soa<PlayerTransform>();
Span<float> x = table.Column<float>(nameof(PlayerTransform.PositionX)).Span;
Span<float> y = table.Column<float>(nameof(PlayerTransform.PositionY)).Span;
@ -170,25 +186,9 @@ for (var row = 0; row < table.Count; row++)
}
```
`ProjectSoa<T>()` is the friendly CPU path: domain documents stay almost-POCO,
while unmanaged public fields/properties become contiguous typed arrays for
DOTS-style scans.
When the locality unit itself needs a durable record key, use
`CultSoaChunk`. A chunk stores entity ids plus named contiguous columns as one
CultCache document:
```csharp
var chunk = CultSoaChunk.Create("players:0", "player.transform.v1", capacity: 1024);
var x = chunk.AddColumn<float>("position.x");
var health = chunk.AddColumn<int>("health");
var row = chunk.AddEntity(entityId);
x.Span[row] = 10;
health.Span[row] = 100;
await cache.UpsertSoaChunkAsync(chunk);
```
`Document<T>()` is the standard presentation. `Soa<T>()` exposes the
cache-owned physical table for hot loops and diagnostics; it is not a second
domain model callers have to maintain.
## Persistence discipline
@ -201,6 +201,8 @@ Useful surfaces:
- `CultCacheMessagePack.OpenAsync(path)`
- `cache.IsDirty`
- `store.IsDirty`
- `cache.Document<T>(key)`
- `cache.Soa<T>()`
- `cache.UpsertAsync(document)`
- `cache.FlushAllBackingStores()`
- `cache.FlushAsync()`

9
src/GameCult.Mesh/CultMesh.cs
View file

@ -238,15 +238,6 @@ namespace GameCult.Mesh
return new CultMeshGameSession(node, options);
}
/// <summary>
/// Creates a SoA chunk store over a CultMesh node database.
/// </summary>
public static CultMeshSoaStore CreateSoaStore(CultMeshNode node)
{
if (node == null) throw new ArgumentNullException(nameof(node));
return new CultMeshSoaStore(node.Database);
}
/// <summary>
/// Attaches Verse discovery responses to a CultMesh node.
/// </summary>

38
src/GameCult.Mesh/CultMeshSoa.cs
View file

@ -1,38 +0,0 @@
using System;
using System.Threading.Tasks;
using GameCult.Caching;
using GameCult.Networking;
namespace GameCult.Mesh
{
/// <summary>
/// Gameplay-facing helper for committing CultCache SoA chunks through CultMesh.
/// </summary>
public sealed class CultMeshSoaStore
{
private readonly CultNetDatabase _database;
internal CultMeshSoaStore(CultNetDatabase database)
{
_database = database ?? throw new ArgumentNullException(nameof(database));
}
/// <summary>
/// Adds or replaces one SoA chunk at a stable mesh record key.
/// </summary>
public Task<CultRecordHandle<CultSoaChunkDocument>> PutChunkAsync(CultRecordKey key, CultSoaChunk chunk)
{
if (chunk == null) throw new ArgumentNullException(nameof(chunk));
return _database.PutAsync(key, chunk.Document);
}
/// <summary>
/// Gets one SoA chunk by record key.
/// </summary>
public async Task<CultSoaChunk?> GetChunkAsync(CultRecordKey key)
{
var document = await _database.GetAsync<CultSoaChunkDocument>(key).ConfigureAwait(false);
return document == null ? null : CultSoaChunk.Wrap(document);
}
}
}

38
src/GameCult.Mesh/README.md
View file

@ -172,34 +172,38 @@ var committed = await session.SubmitAndCommitAsync(new CultNetSimulationObservat
});
```
## CPU-Local SoA Chunks
## Managed Documents
Use regular CultCache documents when that is the clean model. When a simulation
or worker needs a mesh-addressable locality page, `CultSoaChunk` stores entity
ids plus contiguous typed columns as one CultCache document and commits through
the normal mesh database path.
CultMesh presents distributed state as the same managed POCO document shape as
CultCache. Local edits commit through shard authority; remote shard-log or
network events update the same watched value.
```csharp
using GameCult.Caching;
using GameCult.Mesh;
var soa = CultMesh.CreateSoaStore(node);
var key = new CultRecordKey("soa:players:0");
var player = node.Database.Document<PlayerTransform>(new CultRecordKey("player:alice"));
var chunk = CultSoaChunk.Create("players:0", "player.transform.v1", capacity: 1024);
var positionX = chunk.AddColumn<float>("position.x");
var velocityX = chunk.AddColumn<float>("velocity.x");
using var subscription = player.Watch().Subscribe(current => Render(current));
var row = chunk.AddEntity(entityId);
positionX.Span[row] = 4;
velocityX.Span[row] = 0.25f;
await player.ReplaceAsync(new PlayerTransform
{
Name = "Alice",
PositionX = 4,
Health = 100
});
await soa.PutChunkAsync(key, chunk);
player.Value!.Health = 90;
await player.CommitAsync();
```
For pure local hot loops over ordinary typed documents, use
`node.Cache.ProjectSoa<TDocument>()`; chunk documents are for durable or
distributed locality pages, not a requirement imposed on every domain model.
The physical cache table remains SoA for CPU-local scans:
```csharp
Span<float> x = node.Cache.Soa<PlayerTransform>()
.Column<float>(nameof(PlayerTransform.PositionX))
.Span;
```
## Witness Consensus

15
src/GameCult.Networking/CultNetDatabase.cs
View file

@ -686,6 +686,21 @@ namespace GameCult.Networking
return Task.FromResult(_cache.Get<T>(key));
}
/// <summary>
/// Opens a reactive POCO presentation for one distributed CultCache document.
/// </summary>
public CultManagedDocument<T> Document<T>(CultRecordKey key) where T : class
{
ThrowIfDisposed();
return new CultManagedDocument<T>(
key,
() => _cache.Get<T>(key),
async value => { await PutAsync(key, value).ConfigureAwait(false); },
WatchRecord<T>(key)
.Where(change => change.Document != null)
.Select(change => change.Document!));
}
/// <summary>
/// Adds or replaces a document at a specific key.
/// </summary>

67
tests/GameCult.Caching.Tests/BackingStoreTests.cs
View file

@ -7,6 +7,7 @@ using GameCult.Caching;
using GameCult.Caching.MessagePack;
using MessagePack;
using NUnit.Framework;
using R3;
namespace GameCult.Caching.Tests
{
@ -123,62 +124,21 @@ namespace GameCult.Caching.Tests
}
[Test]
public async Task CultCacheSoaChunk_Preserves_Contiguous_Typed_Columns()
{
var filePath = Path.Combine(Path.GetTempPath(), $"cultlib-tests-{Guid.NewGuid():N}.msgpack");
try
{
var cache = await CultCacheMessagePack.OpenAsync(filePath);
var chunk = CultSoaChunk.Create("players:0", "player.transform.v1", capacity: 4);
var positionX = chunk.AddColumn<float>("position.x");
var positionY = chunk.AddColumn<float>("position.y");
var health = chunk.AddColumn<int>("health");
var alice = chunk.AddEntity(101);
var bob = chunk.AddEntity(102);
positionX.Span[alice] = 10.5f;
positionY.Span[alice] = 20.25f;
health.Span[alice] = 90;
positionX.Span[bob] = 11.5f;
positionY.Span[bob] = 21.25f;
health.Span[bob] = 75;
var handle = await cache.UpsertSoaChunkAsync(chunk);
await cache.FlushAsync();
cache.Dispose();
var reopened = await CultCacheMessagePack.OpenAsync(filePath);
var loaded = reopened.GetSoaChunk(handle.Key);
Assert.That(loaded, Is.Not.Null);
Assert.That(loaded!.Count, Is.EqualTo(2));
Assert.That(loaded.Capacity, Is.EqualTo(4));
Assert.That(loaded.ActiveEntityIds.ToArray(), Is.EqualTo(new ulong[] { 101, 102 }));
Assert.That(loaded.Column<float>("position.x").ActiveSpan.ToArray(), Is.EqualTo(new[] { 10.5f, 11.5f }));
Assert.That(loaded.Column<float>("position.y").ActiveSpan.ToArray(), Is.EqualTo(new[] { 20.25f, 21.25f }));
Assert.That(loaded.Column<int>("health").ActiveSpan.ToArray(), Is.EqualTo(new[] { 90, 75 }));
}
finally
{
if (File.Exists(filePath))
{
File.Delete(filePath);
}
}
}
[Test]
public async Task CultCacheSoaProjection_Keeps_Documents_Poco_And_Projects_Hot_Columns()
public async Task CultCacheManagedDocument_Keeps_Poco_Surface_And_Updates_Soa_Storage()
{
var cache = new CultCache();
await cache.UpsertAsync(new TransformEntry
var aliceKey = new CultRecordKey("entity:alice");
var managed = cache.Document<TransformEntry>(aliceKey);
TransformEntry? observed = null;
using var subscription = managed.Watch().Subscribe(value => observed = value);
await managed.ReplaceAsync(new TransformEntry
{
Name = "alice",
PositionX = 1.5f,
PositionY = 2.5f,
Health = 90
}, new CultRecordHandle<TransformEntry>(new CultRecordKey("entity:alice")));
});
await cache.UpsertAsync(new TransformEntry
{
Name = "bob",
@ -187,13 +147,18 @@ namespace GameCult.Caching.Tests
Health = 80
}, new CultRecordHandle<TransformEntry>(new CultRecordKey("entity:bob")));
var table = cache.ProjectSoa<TransformEntry>();
managed.Value!.Health = 70;
await managed.CommitAsync();
var table = cache.Soa<TransformEntry>();
Assert.That(table.Count, Is.EqualTo(2));
Assert.That(table.Keys.Select(key => key.Value).ToArray(), Is.EqualTo(new[] { "entity:alice", "entity:bob" }));
Assert.That(table.Column<float>(nameof(TransformEntry.PositionX)).Span.ToArray(), Is.EqualTo(new[] { 1.5f, 3.5f }));
Assert.That(table.Column<float>(nameof(TransformEntry.PositionY)).Span.ToArray(), Is.EqualTo(new[] { 2.5f, 4.5f }));
Assert.That(table.Column<int>(nameof(TransformEntry.Health)).Span.ToArray(), Is.EqualTo(new[] { 90, 80 }));
Assert.That(table.Column<int>(nameof(TransformEntry.Health)).Span.ToArray(), Is.EqualTo(new[] { 70, 80 }));
Assert.That(observed, Is.Not.Null);
Assert.That(observed!.Health, Is.EqualTo(70));
}
[Test]

57
tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs
View file

@ -4,39 +4,47 @@ using System.Threading.Tasks;
using FluentAssertions;
using GameCult.Caching;
using NUnit.Framework;
using R3;
namespace GameCult.Mesh.Tests;
public sealed class CultMeshStreamingTests
{
[Test]
public async Task SoaStore_Commits_Contiguous_Chunks_Through_MeshDatabase()
public async Task ManagedDocument_Commits_Through_MeshDatabase_And_Watches_Networked_Updates()
{
var filePath = Path.Combine(Path.GetTempPath(), $"cultmesh-soa-{Guid.NewGuid():N}.ccmp");
var filePath = Path.Combine(Path.GetTempPath(), $"cultmesh-managed-{Guid.NewGuid():N}.ccmp");
try
{
using var node = await CultMesh.CreateNodeAsync(
filePath,
new CultMeshNodeOptions { StartServer = false });
var store = CultMesh.CreateSoaStore(node);
var key = new CultRecordKey("soa:players:0");
var chunk = CultSoaChunk.Create("players:0", "player.transform.v1", capacity: 2);
var positionX = chunk.AddColumn<float>("position.x");
var velocityX = chunk.AddColumn<float>("velocity.x");
var key = new CultRecordKey("player:alice");
var document = node.Database.Document<MeshManagedPlayer>(key);
MeshManagedPlayer observed = null!;
using var subscription = document.Watch().Subscribe(value => observed = value);
var row = chunk.AddEntity(5001);
positionX.Span[row] = 4;
velocityX.Span[row] = 0.25f;
await document.ReplaceAsync(new MeshManagedPlayer
{
Name = "alice",
PositionX = 4,
Health = 100
});
await node.Database.PutAsync(key, new MeshManagedPlayer
{
Name = "alice",
PositionX = 8,
Health = 75
});
await store.PutChunkAsync(key, chunk);
var loaded = await store.GetChunkAsync(key);
loaded.Should().NotBeNull();
loaded!.ActiveEntityIds.ToArray().Should().Equal(5001UL);
loaded.Column<float>("position.x").ActiveSpan.ToArray().Should().Equal(4);
loaded.Column<float>("velocity.x").ActiveSpan.ToArray().Should().Equal(0.25f);
document.Value.Should().NotBeNull();
document.Value!.Health.Should().Be(75);
observed.Should().NotBeNull();
observed!.PositionX.Should().Be(8);
node.Cache.Soa<MeshManagedPlayer>().Column<int>(nameof(MeshManagedPlayer.Health)).Span.ToArray()
.Should()
.Equal(75);
}
finally
{
@ -200,4 +208,17 @@ public sealed class CultMeshStreamingTests
new[] { CultMeshStreamBodyTransport.SharedMemory, CultMeshStreamBodyTransport.CultCachePage }));
return catalog;
}
[CultDocument("tests.mesh_managed_player", "tests.mesh_managed_player.v1")]
private sealed class MeshManagedPlayer
{
[MessagePack.Key(0)]
public string Name = string.Empty;
[MessagePack.Key(1)]
public float PositionX;
[MessagePack.Key(2)]
public int Health;
}
}