diff --git a/src/GameCult.Caching/CultCache.cs b/src/GameCult.Caching/CultCache.cs index faaa30a0..e5e1d31a 100644 --- a/src/GameCult.Caching/CultCache.cs +++ b/src/GameCult.Caching/CultCache.cs @@ -1159,6 +1159,8 @@ namespace GameCult.Caching private readonly ConcurrentDictionary<(Type Type, string Alias), ConcurrentDictionary> _indexMaps = new(); private readonly ConcurrentDictionary _globalKeys = new(); private readonly ConditionalWeakTable _documentHandles = new(); + private readonly Subject _changes = new(); + private readonly CultCacheSoaStore _soa = new(); private ILogger _logger = new NullLogger(); private bool _hasUnflushedMutations; @@ -1222,6 +1224,46 @@ namespace GameCult.Caching /// public CultDocumentRegistry Registry => _registry; + /// + /// Opens a reactive POCO presentation for one cache-managed document. + /// + public CultManagedDocument Document(CultRecordKey key) where T : class + { + return new CultManagedDocument( + key, + () => Get(key), + async value => { await UpsertAsync(value, new CultRecordHandle(key)).ConfigureAwait(false); }, + WatchRecord(key) + .Where(change => change.Document != null) + .Select(change => change.Document!)); + } + + /// + /// Watches all local cache changes assignable to the requested document type. + /// + public Observable> Watch() where T : class + { + return _changes + .Where(change => change is CultCacheDocumentChange) + .Select(change => (CultCacheDocumentChange)change); + } + + /// + /// Watches one local cache record. + /// + public Observable> WatchRecord(CultRecordKey key) where T : class + { + return Watch().Where(change => change.Key.Equals(key)); + } + + /// + /// Gets the cache-owned structure-of-arrays table for a document type. + /// + public CultSoaTable Soa() where T : class + { + return _soa.Snapshot(); + } + /// /// Attaches a backing store to this cache. /// @@ -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); + } + } } /// diff --git a/src/GameCult.Caching/CultManagedDocument.cs b/src/GameCult.Caching/CultManagedDocument.cs new file mode 100644 index 00000000..100cb628 --- /dev/null +++ b/src/GameCult.Caching/CultManagedDocument.cs @@ -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 +{ + /// + /// Describes how a CultCache-managed document changed. + /// + public enum CultCacheDocumentChangeKind + { + /// + /// A document was added. + /// + Added, + + /// + /// A document was updated. + /// + Updated, + + /// + /// A document was removed. + /// + Removed + } + + /// + /// One reactive CultCache document change. + /// + public sealed class CultCacheDocumentChange where T : class + { + /// + /// Creates a document change. + /// + public CultCacheDocumentChange( + CultCacheDocumentChangeKind kind, + CultRecordKey key, + T? document, + T? previousDocument) + { + Kind = kind; + Key = key; + Document = document; + PreviousDocument = previousDocument; + } + + /// + /// Gets the change kind. + /// + public CultCacheDocumentChangeKind Kind { get; } + + /// + /// Gets the document record key. + /// + public CultRecordKey Key { get; } + + /// + /// Gets the current POCO presentation, when present. + /// + public T? Document { get; } + + /// + /// Gets the previous POCO presentation, when present. + /// + public T? PreviousDocument { get; } + } + + /// + /// Reactive POCO presentation for one CultCache-managed document. + /// + public sealed class CultManagedDocument where T : class + { + private readonly Func _read; + private readonly Func _commit; + private readonly Observable _watch; + + /// + /// Creates a managed document presentation. + /// + public CultManagedDocument( + CultRecordKey key, + Func read, + Func commit, + Observable watch) + { + Key = key; + _read = read ?? throw new ArgumentNullException(nameof(read)); + _commit = commit ?? throw new ArgumentNullException(nameof(commit)); + _watch = watch ?? throw new ArgumentNullException(nameof(watch)); + } + + /// + /// Gets the document record key. + /// + public CultRecordKey Key { get; } + + /// + /// Gets the current POCO presentation. + /// + public T? Value => _read(); + + /// + /// Watches committed values for this document. + /// + public Observable Watch() + { + return _watch; + } + + /// + /// Commits the current POCO value through the owner-supplied commit path. + /// + public Task CommitAsync() + { + var value = Value ?? throw new InvalidOperationException($"Document '{Key.Value}' is not present."); + return _commit(value); + } + + /// + /// Replaces the current value through the owner-supplied commit path. + /// + public Task ReplaceAsync(T value) + { + if (value == null) throw new ArgumentNullException(nameof(value)); + return _commit(value); + } + } + + /// + /// CPU-local structure-of-arrays table owned by CultCache for one document type. + /// + public sealed class CultSoaTable where TDocument : class + { + private readonly Dictionary _columns; + + internal CultSoaTable( + CultRecordKey[] keys, + TDocument[] documents, + Dictionary columns) + { + Keys = keys; + Documents = documents; + _columns = columns; + } + + /// + /// Gets record keys in row order. + /// + public IReadOnlyList Keys { get; } + + /// + /// Gets POCO presentations in row order. + /// + public IReadOnlyList Documents { get; } + + /// + /// Gets the row count. + /// + public int Count => Keys.Count; + + /// + /// Gets a contiguous unmanaged field/property column. + /// + public CultSoaColumn Column(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(name, values); + } + } + + /// + /// Contiguous typed SoA column. + /// + public readonly struct CultSoaColumn where TValue : unmanaged + { + internal CultSoaColumn(string name, TValue[] values) + { + Name = name; + Values = values; + } + + /// + /// Gets the column name. + /// + public string Name { get; } + + /// + /// Gets the backing array. + /// + public TValue[] Values { get; } + + /// + /// Gets the backing array as a span. + /// + public Span Span => Values.AsSpan(); + } + + internal sealed class CultCacheSoaStore + { + private readonly Dictionary _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 Snapshot() where TDocument : class + { + return Table(typeof(TDocument)).Snapshot(); + } + + 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 _keys = new(); + private readonly List _documents = new(); + private readonly Dictionary _members; + private readonly Dictionary _columns; + private readonly Dictionary _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 Snapshot() 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(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( + _keys.ToArray(), + _documents.Cast().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 getter) + { + Name = name; + ValueType = valueType; + Get = getter; + } + + public string Name { get; } + public Type ValueType { get; } + public Func Get { get; } + + public static IEnumerable 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); + } + } +} diff --git a/src/GameCult.Caching/CultSoa.cs b/src/GameCult.Caching/CultSoa.cs deleted file mode 100644 index 5abd53bd..00000000 --- a/src/GameCult.Caching/CultSoa.cs +++ /dev/null @@ -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 -{ - /// - /// CultCache document that stores a CPU-local structure-of-arrays chunk. - /// - [CultDocument("gamecult.soa_chunk", "gamecult.soa_chunk.v1")] - public sealed class CultSoaChunkDocument - { - /// - /// Gets or sets the stable chunk identifier. - /// - [CultName] - public string ChunkId { get; set; } = string.Empty; - - /// - /// Gets or sets the layout/archetype identifier shared by chunks with the same columns. - /// - [CultIndex("archetype")] - public string ArchetypeId { get; set; } = string.Empty; - - /// - /// Gets or sets the number of live rows in this chunk. - /// - public int Count { get; set; } - - /// - /// Gets or sets the live row entity ids. The array length is the chunk capacity. - /// - public ulong[] EntityIds { get; set; } = Array.Empty(); - - /// - /// Gets or sets the contiguous component columns owned by this chunk. - /// - public CultSoaColumnDocument[] Columns { get; set; } = Array.Empty(); - } - - /// - /// One contiguous column inside a structure-of-arrays chunk. - /// - public sealed class CultSoaColumnDocument - { - /// - /// Gets or sets the stable column name, usually a component field path. - /// - public string Name { get; set; } = string.Empty; - - /// - /// Gets or sets the CLR element type name stored in the column. - /// - public string ElementType { get; set; } = string.Empty; - - /// - /// Gets or sets the size of one element in bytes. - /// - public int ElementByteLength { get; set; } - - /// - /// Gets or sets the contiguous column payload. - /// - public byte[] Data { get; set; } = Array.Empty(); - } - - /// - /// Runtime view over a CultCache SoA chunk document. - /// - public sealed class CultSoaChunk - { - private readonly Dictionary _columns; - - private CultSoaChunk(CultSoaChunkDocument document) - { - Document = document ?? throw new ArgumentNullException(nameof(document)); - ValidateDocument(document); - _columns = document.Columns.ToDictionary(column => column.Name, StringComparer.Ordinal); - } - - /// - /// Gets the underlying CultCache document. - /// - public CultSoaChunkDocument Document { get; } - - /// - /// Gets the stable chunk identifier. - /// - public string ChunkId => Document.ChunkId; - - /// - /// Gets the layout/archetype identifier. - /// - public string ArchetypeId => Document.ArchetypeId; - - /// - /// Gets the number of live rows. - /// - public int Count => Document.Count; - - /// - /// Gets the maximum row count this chunk can hold without reallocating. - /// - public int Capacity => Document.EntityIds.Length; - - /// - /// Gets all entity ids, including unused capacity. - /// - public Span EntityIds => Document.EntityIds.AsSpan(); - - /// - /// Gets live entity ids. - /// - public Span ActiveEntityIds => Document.EntityIds.AsSpan(0, Document.Count); - - /// - /// Creates an empty SoA chunk with the supplied capacity. - /// - 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() - }); - } - - /// - /// Wraps a persisted SoA chunk document. - /// - public static CultSoaChunk Wrap(CultSoaChunkDocument document) - { - return new CultSoaChunk(document); - } - - /// - /// Sets the live row count after validating all columns have enough capacity. - /// - public void SetCount(int count) - { - if (count < 0 || count > Capacity) - { - throw new ArgumentOutOfRangeException(nameof(count)); - } - - Document.Count = count; - } - - /// - /// Appends one entity row and returns its row index. - /// - 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; - } - - /// - /// Adds a contiguous typed column. - /// - public CultSoaColumn AddColumn(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(), - ElementByteLength = Marshal.SizeOf(), - Data = new byte[checked(Capacity * Marshal.SizeOf())] - }; - - var columns = Document.Columns.ToList(); - columns.Add(column); - Document.Columns = columns.ToArray(); - _columns[name] = column; - return new CultSoaColumn(this, column); - } - - /// - /// Gets an existing typed column. - /// - public CultSoaColumn Column(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(); - var expectedByteLength = Marshal.SizeOf(); - 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(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(); - document.Columns ??= Array.Empty(); - 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(); - 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)); - } - } - } - } - - /// - /// Typed contiguous view over one SoA column. - /// - public readonly struct CultSoaColumn where T : unmanaged - { - private readonly CultSoaChunk _owner; - private readonly CultSoaColumnDocument _column; - - internal CultSoaColumn(CultSoaChunk owner, CultSoaColumnDocument column) - { - _owner = owner; - _column = column; - } - - /// - /// Gets the column name. - /// - public string Name => _column.Name; - - /// - /// Gets the full column capacity as a typed span. - /// - public Span Span => MemoryMarshal.Cast(_column.Data.AsSpan()); - - /// - /// Gets the live rows as a typed span. - /// - public Span ActiveSpan => Span.Slice(0, _owner.Count); - } - - /// - /// Convenience extensions for storing SoA chunks in CultCache. - /// - public static class CultSoaCacheExtensions - { - /// - /// Adds or replaces a SoA chunk document in a CultCache. - /// - public static Task> UpsertSoaChunkAsync( - this CultCache cache, - CultSoaChunk chunk, - CultRecordHandle? handle = null) - { - if (cache == null) throw new ArgumentNullException(nameof(cache)); - if (chunk == null) throw new ArgumentNullException(nameof(chunk)); - return cache.UpsertAsync(chunk.Document, handle); - } - - /// - /// Gets a SoA chunk view by record key. - /// - public static CultSoaChunk? GetSoaChunk(this CultCache cache, CultRecordKey key) - { - if (cache == null) throw new ArgumentNullException(nameof(cache)); - var document = cache.Get(key); - return document == null ? null : CultSoaChunk.Wrap(document); - } - - /// - /// Projects ordinary typed CultCache documents into contiguous unmanaged columns for hot CPU loops. - /// - public static CultSoaDocumentTable ProjectSoa(this CultCache cache) where TDocument : class - { - if (cache == null) throw new ArgumentNullException(nameof(cache)); - return CultSoaDocumentTable.From(cache); - } - } - - /// - /// Structure-of-arrays projection over ordinary CultCache documents. - /// - public sealed class CultSoaDocumentTable where TDocument : class - { - private readonly Dictionary _columns; - - private CultSoaDocumentTable( - CultRecordKey[] keys, - TDocument[] documents, - Dictionary columns) - { - Keys = keys; - Documents = documents; - _columns = columns; - } - - /// - /// Gets the projected record keys in row order. - /// - public IReadOnlyList Keys { get; } - - /// - /// Gets the projected document references in row order. - /// - public IReadOnlyList Documents { get; } - - /// - /// Gets the row count. - /// - public int Count => Documents.Count; - - /// - /// Builds a contiguous projection from all documents of this type in the cache. - /// - public static CultSoaDocumentTable 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(StringComparer.Ordinal); - - foreach (var member in CultSoaDocumentMember.Discover(typeof(TDocument))) - { - columns[member.Name] = member.Project(documents); - } - - return new CultSoaDocumentTable(keys, documents, columns); - } - - /// - /// Gets a contiguous projected column by field or property name. - /// - public CultSoaDocumentColumn Column(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(name, values); - } - } - - /// - /// Contiguous projected column from ordinary CultCache documents. - /// - public readonly struct CultSoaDocumentColumn where TValue : unmanaged - { - internal CultSoaDocumentColumn(string name, TValue[] values) - { - Name = name; - Values = values; - } - - /// - /// Gets the projected field or property name. - /// - public string Name { get; } - - /// - /// Gets the backing array. - /// - public TValue[] Values { get; } - - /// - /// Gets the column as a contiguous span. - /// - public Span Span => Values.AsSpan(); - } - - internal sealed class CultSoaDocumentMember - { - private CultSoaDocumentMember(string name, Type valueType, Func getter) - { - Name = name; - ValueType = valueType; - Getter = getter; - } - - public string Name { get; } - public Type ValueType { get; } - private Func Getter { get; } - - public static IEnumerable 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 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() 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; - } - } - - /// - /// Built-in metadata and payload codec for CultCache SoA documents. - /// - public sealed class CultSoaGeneratedDocumentMetadataProvider : ICultGeneratedDocumentMetadataProvider - { - /// - public IEnumerable 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(); - writer.Write(entityIds.Length); - foreach (var entityId in entityIds) - { - writer.Write(entityId); - } - - var columns = chunk.Columns ?? Array.Empty(); - 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(); - 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; - } - } -} diff --git a/src/GameCult.Caching/README.md b/src/GameCult.Caching/README.md index c0e9270a..7e51227b 100644 --- a/src/GameCult.Caching/README.md +++ b/src/GameCult.Caching/README.md @@ -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(); +var player = cache.Document(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(); Span x = table.Column(nameof(PlayerTransform.PositionX)).Span; Span y = table.Column(nameof(PlayerTransform.PositionY)).Span; @@ -170,25 +186,9 @@ for (var row = 0; row < table.Count; row++) } ``` -`ProjectSoa()` 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("position.x"); -var health = chunk.AddColumn("health"); - -var row = chunk.AddEntity(entityId); -x.Span[row] = 10; -health.Span[row] = 100; - -await cache.UpsertSoaChunkAsync(chunk); -``` +`Document()` is the standard presentation. `Soa()` 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(key)` +- `cache.Soa()` - `cache.UpsertAsync(document)` - `cache.FlushAllBackingStores()` - `cache.FlushAsync()` diff --git a/src/GameCult.Mesh/CultMesh.cs b/src/GameCult.Mesh/CultMesh.cs index 3a0bc51c..8cbeec76 100644 --- a/src/GameCult.Mesh/CultMesh.cs +++ b/src/GameCult.Mesh/CultMesh.cs @@ -238,15 +238,6 @@ namespace GameCult.Mesh return new CultMeshGameSession(node, options); } - /// - /// Creates a SoA chunk store over a CultMesh node database. - /// - public static CultMeshSoaStore CreateSoaStore(CultMeshNode node) - { - if (node == null) throw new ArgumentNullException(nameof(node)); - return new CultMeshSoaStore(node.Database); - } - /// /// Attaches Verse discovery responses to a CultMesh node. /// diff --git a/src/GameCult.Mesh/CultMeshSoa.cs b/src/GameCult.Mesh/CultMeshSoa.cs deleted file mode 100644 index 5ea00c7c..00000000 --- a/src/GameCult.Mesh/CultMeshSoa.cs +++ /dev/null @@ -1,38 +0,0 @@ -using System; -using System.Threading.Tasks; -using GameCult.Caching; -using GameCult.Networking; - -namespace GameCult.Mesh -{ - /// - /// Gameplay-facing helper for committing CultCache SoA chunks through CultMesh. - /// - public sealed class CultMeshSoaStore - { - private readonly CultNetDatabase _database; - - internal CultMeshSoaStore(CultNetDatabase database) - { - _database = database ?? throw new ArgumentNullException(nameof(database)); - } - - /// - /// Adds or replaces one SoA chunk at a stable mesh record key. - /// - public Task> PutChunkAsync(CultRecordKey key, CultSoaChunk chunk) - { - if (chunk == null) throw new ArgumentNullException(nameof(chunk)); - return _database.PutAsync(key, chunk.Document); - } - - /// - /// Gets one SoA chunk by record key. - /// - public async Task GetChunkAsync(CultRecordKey key) - { - var document = await _database.GetAsync(key).ConfigureAwait(false); - return document == null ? null : CultSoaChunk.Wrap(document); - } - } -} diff --git a/src/GameCult.Mesh/README.md b/src/GameCult.Mesh/README.md index 0cd735f9..72f286ff 100644 --- a/src/GameCult.Mesh/README.md +++ b/src/GameCult.Mesh/README.md @@ -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(new CultRecordKey("player:alice")); -var chunk = CultSoaChunk.Create("players:0", "player.transform.v1", capacity: 1024); -var positionX = chunk.AddColumn("position.x"); -var velocityX = chunk.AddColumn("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()`; 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 x = node.Cache.Soa() + .Column(nameof(PlayerTransform.PositionX)) + .Span; +``` ## Witness Consensus diff --git a/src/GameCult.Networking/CultNetDatabase.cs b/src/GameCult.Networking/CultNetDatabase.cs index 93b91d44..83c34b82 100644 --- a/src/GameCult.Networking/CultNetDatabase.cs +++ b/src/GameCult.Networking/CultNetDatabase.cs @@ -686,6 +686,21 @@ namespace GameCult.Networking return Task.FromResult(_cache.Get(key)); } + /// + /// Opens a reactive POCO presentation for one distributed CultCache document. + /// + public CultManagedDocument Document(CultRecordKey key) where T : class + { + ThrowIfDisposed(); + return new CultManagedDocument( + key, + () => _cache.Get(key), + async value => { await PutAsync(key, value).ConfigureAwait(false); }, + WatchRecord(key) + .Where(change => change.Document != null) + .Select(change => change.Document!)); + } + /// /// Adds or replaces a document at a specific key. /// diff --git a/tests/GameCult.Caching.Tests/BackingStoreTests.cs b/tests/GameCult.Caching.Tests/BackingStoreTests.cs index 22e1b8b2..e08a2ebc 100644 --- a/tests/GameCult.Caching.Tests/BackingStoreTests.cs +++ b/tests/GameCult.Caching.Tests/BackingStoreTests.cs @@ -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("position.x"); - var positionY = chunk.AddColumn("position.y"); - var health = chunk.AddColumn("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("position.x").ActiveSpan.ToArray(), Is.EqualTo(new[] { 10.5f, 11.5f })); - Assert.That(loaded.Column("position.y").ActiveSpan.ToArray(), Is.EqualTo(new[] { 20.25f, 21.25f })); - Assert.That(loaded.Column("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(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(new CultRecordKey("entity:alice"))); + }); await cache.UpsertAsync(new TransformEntry { Name = "bob", @@ -187,13 +147,18 @@ namespace GameCult.Caching.Tests Health = 80 }, new CultRecordHandle(new CultRecordKey("entity:bob"))); - var table = cache.ProjectSoa(); + managed.Value!.Health = 70; + await managed.CommitAsync(); + + var table = cache.Soa(); 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(nameof(TransformEntry.PositionX)).Span.ToArray(), Is.EqualTo(new[] { 1.5f, 3.5f })); Assert.That(table.Column(nameof(TransformEntry.PositionY)).Span.ToArray(), Is.EqualTo(new[] { 2.5f, 4.5f })); - Assert.That(table.Column(nameof(TransformEntry.Health)).Span.ToArray(), Is.EqualTo(new[] { 90, 80 })); + Assert.That(table.Column(nameof(TransformEntry.Health)).Span.ToArray(), Is.EqualTo(new[] { 70, 80 })); + Assert.That(observed, Is.Not.Null); + Assert.That(observed!.Health, Is.EqualTo(70)); } [Test] diff --git a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs index a1888897..4278d621 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs @@ -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("position.x"); - var velocityX = chunk.AddColumn("velocity.x"); + var key = new CultRecordKey("player:alice"); + var document = node.Database.Document(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("position.x").ActiveSpan.ToArray().Should().Equal(4); - loaded.Column("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().Column(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; + } }