Added performance test project and lots of improvements

src/net/Common/Common.props

@@ -4,7 +4,7 @@
 		<Owners>MASES s.r.l.</Owners>
 		<Authors>MASES s.r.l.</Authors>
 		<Company>MASES s.r.l.</Company>
-		<Version>0.8.1.0</Version>
+		<Version>0.8.2.0</Version>
 		<TargetFrameworks>net6.0</TargetFrameworks>
 		<LangVersion>latest</LangVersion>
 		<GenerateAssemblyInfo>true</GenerateAssemblyInfo>

src/net/KEFCore/Infrastructure/KafkaDbContext.cs

@@ -19,7 +19,6 @@
 *  Refer to LICENSE for more information.
 */
 
-using MASES.KNet;
 using MASES.KNet.Common;
 using MASES.KNet.Producer;
 using MASES.KNet.Streams;
@@ -31,6 +30,50 @@ namespace MASES.EntityFrameworkCore.KNet.Infrastructure;
 /// </summary>
 public class KafkaDbContext : DbContext
 {
+#if DEBUG_PERFORMANCE
+    const bool perf = true;
+    /// <summary>
+    /// Enable tracing of <see cref="MASES.EntityFrameworkCore.KNet.Storage.Internal.EntityTypeDataStorage{TKey}"/>
+    /// </summary>
+    public static bool TraceEntityTypeDataStorageGetData = false;
+
+    public static void ReportString(string message)
+    {
+        if (!_enableKEFCoreTracing) return;
+
+        if (Debugger.IsAttached)
+        {
+            Trace.WriteLine($"{DateTime.Now:HH::mm::ss:ffff} - {message}");
+        }
+        else
+        {
+            Console.WriteLine($"{DateTime.Now:HH::mm::ss:ffff} - {message}");
+        }
+    }
+#else
+const bool perf = false;
+#endif
+    /// <summary>
+    /// Reports if the library was compiled to reports performance information
+    /// </summary>
+    public const bool IsPerformanceVersion = perf;
+    static bool _enableKEFCoreTracing = false;
+    /// <summary>
+    /// Set to <see langword="true"/> to enable tracing of KEFCore
+    /// </summary>
+    /// <remarks>Can be set only if the project is compiled with DEBUG_PERFORMANCE preprocessor directive, otherwise an <see cref="InvalidOperationException"/> is raised</remarks>
+    public static bool EnableKEFCoreTracing
+    {
+        get { return _enableKEFCoreTracing; }
+        set 
+        {
+            _enableKEFCoreTracing = value;
+#if DEBUG_PERFORMANCE
+            if (_enableKEFCoreTracing) throw new InvalidOperationException("Compile KEFCore using DEBUG_PERFORMANCE preprocessor directive");
+#endif
+        }
+    }
+
     /// <inheritdoc cref="DbContext.DbContext()"/>
     public KafkaDbContext()
     {

src/net/KEFCore/KEFCore.csproj

@@ -66,11 +66,11 @@
 	</ItemGroup>
 
 	<ItemGroup>
-		<PackageReference Include="MASES.KNet" Version="2.1.2">
+		<PackageReference Include="MASES.KNet" Version="2.1.3">
 			<IncludeAssets>All</IncludeAssets>
 			<PrivateAssets>None</PrivateAssets>
 		</PackageReference>
-		<PackageReference Include="MASES.KNet.Serialization.Json" Version="2.1.2" />
-		<PackageReference Include="Microsoft.EntityFrameworkCore" Version="7.0.11" PrivateAssets="none" />
+		<PackageReference Include="MASES.KNet.Serialization.Json" Version="2.1.3" />
+		<PackageReference Include="Microsoft.EntityFrameworkCore" Version="7.0.12" PrivateAssets="none" />
 	</ItemGroup>
 </Project>

src/net/KEFCore/Query/Internal/KafkaQueryContext.cs

@@ -25,26 +25,16 @@ namespace MASES.EntityFrameworkCore.KNet.Query.Internal;
 
 public class KafkaQueryContext : QueryContext
 {
-    private readonly IDictionary<IEntityType, IEnumerable<ValueBuffer>> _valueBuffersCache
-        = new Dictionary<IEntityType, IEnumerable<ValueBuffer>>();
+    private readonly IKafkaCluster _cluster;
 
     public virtual IEnumerable<ValueBuffer> GetValueBuffers(IEntityType entityType)
     {
-        if (!_valueBuffersCache.TryGetValue(entityType, out var valueBuffers))
-        {
-            valueBuffers = Cluster.GetValueBuffers(entityType);
-
-            _valueBuffersCache[entityType] = valueBuffers;
-        }
-
-        return valueBuffers;
+        return _cluster.GetValueBuffers(entityType);
     }
 
     public KafkaQueryContext(QueryContextDependencies dependencies, IKafkaCluster cluster)
         : base(dependencies)
     {
-        Cluster = cluster;
+        _cluster = cluster;
     }
-
-    public virtual IKafkaCluster Cluster { get; }
 }

src/net/KEFCore/Query/Internal/KafkaQueryContextFactory.cs

@@ -26,7 +26,6 @@ namespace MASES.EntityFrameworkCore.KNet.Query.Internal;
 public class KafkaQueryContextFactory : IQueryContextFactory
 {
     private readonly IKafkaCluster _cluster;
-    readonly KafkaQueryContext context;
 
     public KafkaQueryContextFactory(
         QueryContextDependencies dependencies,
@@ -35,13 +34,12 @@ public KafkaQueryContextFactory(
     {
         _cluster = clusterCache.GetCluster(contextOptions);
         Dependencies = dependencies;
-        context = new KafkaQueryContext(Dependencies, _cluster);
     }
 
     /// <summary>
     ///     Dependencies for this service.
     /// </summary>
     protected virtual QueryContextDependencies Dependencies { get; }
 
-    public virtual QueryContext Create() => context; // new KafkaQueryContext(Dependencies, _cluster);
+    public virtual QueryContext Create() => new KafkaQueryContext(Dependencies, _cluster);
 }

src/net/KEFCore/Storage/Internal/EntityTypeDataStorage.cs

@@ -16,6 +16,8 @@
 *  Refer to LICENSE for more information.
 */
 
+// #define DEBUG_PERFORMANCE
+
 #nullable enable
 
 using System.Text.Json;
@@ -167,7 +169,10 @@ public void GetData(IEntityType tName, ref object[] array)
         }
         finally
         {
-            Trace.WriteLine($"Time to GetData with length {Data.Count}: {fullSw.Elapsed} - new array took: {newSw.Elapsed} - Iteration took: {iterationSw.Elapsed}");
+            if (Infrastructure.KafkaDbContext.TraceEntityTypeDataStorageGetData)
+            {
+                Infrastructure.KafkaDbContext.ReportString($"Time to GetData with length {Data.Count}: {fullSw.Elapsed} - new array took: {newSw.Elapsed} - Iteration took: {iterationSw.Elapsed}");
+            }
         }
 #endif
     }

src/net/KEFCore/Storage/Internal/EntityTypeProducer.cs

@@ -50,37 +50,99 @@
         #region KNetCompactedReplicatorEnumerator
         class KNetCompactedReplicatorEnumerator : IEnumerator<ValueBuffer>
         {
+#if DEBUG_PERFORMANCE
+            Stopwatch _moveNextSw = new Stopwatch();
+            Stopwatch _currentSw = new Stopwatch();
+            Stopwatch _valueBufferSw = new Stopwatch();
+#endif
             readonly IEntityType _entityType;
+            IKNetCompactedReplicator<TKey, EntityTypeDataStorage<TKey>>? _kafkaCompactedReplicator;
             readonly IEnumerator<KeyValuePair<TKey, EntityTypeDataStorage<TKey>>> _enumerator;
             public KNetCompactedReplicatorEnumerator(IEntityType entityType, IKNetCompactedReplicator<TKey, EntityTypeDataStorage<TKey>>? kafkaCompactedReplicator)
             {
                 _entityType = entityType;
-                kafkaCompactedReplicator?.SyncWait();
-                _enumerator = kafkaCompactedReplicator?.GetEnumerator();
+                _kafkaCompactedReplicator = kafkaCompactedReplicator;
+#if DEBUG_PERFORMANCE
+                Stopwatch sw = Stopwatch.StartNew();
+#endif
+                if (!_kafkaCompactedReplicator!.SyncWait()) throw new InvalidOperationException($"Failed to synchronize with {_kafkaCompactedReplicator.StateName}");
+#if DEBUG_PERFORMANCE
+                sw.Stop();
+                Infrastructure.KafkaDbContext.ReportString($"KNetCompactedReplicatorEnumerator SyncWait for {_entityType.Name} tooks {sw.Elapsed}");
+#endif
+                _enumerator = _kafkaCompactedReplicator?.GetEnumerator();
             }
 
             ValueBuffer? _current = null;
 
-            public ValueBuffer Current => _current.HasValue ? _current.Value : default;
+            public ValueBuffer Current
+            {
+                get
+                {
+#if DEBUG_PERFORMANCE
+                    try
+                    {
+                        _currentSw.Start();
+#endif
+                        return _current.HasValue ? _current.Value : default;
+#if DEBUG_PERFORMANCE
+                    }
+                    finally
+                    {
+                        _currentSw.Stop();
+                    }
+#endif
+                }
+            }
 
             object IEnumerator.Current => Current;
 
             public void Dispose()
             {
+#if DEBUG_PERFORMANCE
+                Infrastructure.KafkaDbContext.ReportString($"KNetCompactedReplicatorEnumerator _moveNextSw: {_moveNextSw.Elapsed} _currentSw: {_currentSw.Elapsed} _valueBufferSw: {_valueBufferSw.Elapsed}");
+#endif
                 _enumerator?.Dispose();
             }
 
+#if DEBUG_PERFORMANCE
+            int _cycles = 0;
+#endif
+
             public bool MoveNext()
             {
-                if (_enumerator.MoveNext())
+#if DEBUG_PERFORMANCE
+                try
+                {
+                    _moveNextSw.Start();
+#endif
+                    if (_enumerator.MoveNext())
+                    {
+#if DEBUG_PERFORMANCE
+                        _cycles++;
+                        _valueBufferSw.Start();
+#endif
+                        object[] array = null;
+                        _enumerator.Current.Value.GetData(_entityType, ref array);
+#if DEBUG_PERFORMANCE
+                        _valueBufferSw.Stop();
+#endif
+                        _current = new ValueBuffer(array);
+                        return true;
+                    }
+                    _current = null;
+                    return false;
+#if DEBUG_PERFORMANCE
+                }
+                finally
                 {
-                    object[] array = null;
-                    _enumerator.Current.Value.GetData(_entityType, ref array);
-                    _current = new ValueBuffer(array);
-                    return true;
+                    _moveNextSw.Stop();
+                    if (_cycles == 0)
+                    {
+                        throw new InvalidOperationException($"KNetCompactedReplicatorEnumerator - No data returned from {_kafkaCompactedReplicator}");
+                    }
                 }
-                _current = null;
-                return false;
+#endif
             }
 
             public void Reset()
@@ -108,8 +170,11 @@
     }
     #endregion
 
     public EntityTypeProducer(IEntityType entityType, IKafkaCluster cluster)
     {
+#if DEBUG_PERFORMANCE
+        Infrastructure.KafkaDbContext.ReportString($"Creating new EntityTypeProducer for {entityType.Name}");
+#endif
         _entityType = entityType;
         _cluster = cluster;
         _useCompactedReplicator = _cluster.Options.UseCompactedReplicator;
@@ -137,7 +202,14 @@
                 KeySerDes = _keySerdes,
                 ValueSerDes = _valueSerdes,
             };
+#if DEBUG_PERFORMANCE
+            Stopwatch sw = Stopwatch.StartNew();
+#endif
             if (!_kafkaCompactedReplicator.StartAndWait()) throw new InvalidOperationException($"Failed to synchronize with {_kafkaCompactedReplicator.StateName}");
+#if DEBUG_PERFORMANCE
+            sw.Stop();
+            Infrastructure.KafkaDbContext.ReportString($"EntityTypeProducer - KNetCompactedReplicator::StartAndWait for {entityType.Name} in {sw.Elapsed}");
+#endif
         }
         else
         {
@@ -146,6 +218,8 @@
         }
     }
 
+    public virtual IEntityType EntityType => _entityType;
+
     public IEnumerable<Future<RecordMetadata>> Commit(IEnumerable<IKafkaRowBag> records)
     {
         if (_useCompactedReplicator)
@@ -156,7 +230,7 @@
                 if (_kafkaCompactedReplicator != null) _kafkaCompactedReplicator[record.Key] = value!;
             }
 
             return null;
         }
         else
         {
@@ -164,7 +238,7 @@
             foreach (KafkaRowBag<TKey> record in records)
             {
                 var future = _kafkaProducer?.Send(new KNetProducerRecord<TKey, EntityTypeDataStorage<TKey>>(record.AssociatedTopicName, 0, record.Key, record.Value!));
                 futures.Add(future);
             }
 
             _kafkaProducer?.Flush();

src/net/KEFCore/Storage/Internal/EntityTypeProducers.cs

@@ -34,18 +34,16 @@ public class EntityTypeProducers
 
     public static IEntityTypeProducer Create<TKey>(IEntityType entityType, IKafkaCluster cluster) where TKey : notnull
     {
-        //if (!cluster.Options.ProducerByEntity)
-        //{
-        //    lock (_producers)
-        //    {
-        //        if (_globalProducer == null) _globalProducer = CreateProducerLocal<TKey>(entityType, cluster);
-        //        return _globalProducer;
-        //    }
-        //}
-        //else
-        //{
         return _producers.GetOrAdd(entityType, _ => CreateProducerLocal<TKey>(entityType, cluster));
-        //}
+    }
+
+    public static void Dispose(IEntityTypeProducer producer)
+    {
+        if (!_producers.TryRemove(new KeyValuePair<IEntityType, IEntityTypeProducer>(producer.EntityType, producer)))
+        {
+            throw new InvalidOperationException($"Failed to remove IEntityTypeProducer for {producer.EntityType.Name}");
+        }
+        producer.Dispose();
     }
 
     static IEntityTypeProducer CreateProducerLocal<TKey>(IEntityType entityType, IKafkaCluster cluster) where TKey : notnull => new EntityTypeProducer<TKey>(entityType, cluster);

src/net/KEFCore/Storage/Internal/IEntityTypeProducer.cs

@@ -25,6 +25,8 @@ namespace MASES.EntityFrameworkCore.KNet.Storage.Internal;
 
 public interface IEntityTypeProducer : IDisposable
 {
+    IEntityType EntityType { get; }
+
     IEnumerable<Future<RecordMetadata>> Commit(IEnumerable<IKafkaRowBag> records);
 
     IEnumerable<ValueBuffer> ValueBuffers { get; }

src/net/KEFCore/Storage/Internal/IKafkaCluster.cs

@@ -37,5 +37,7 @@ public interface IKafkaCluster : IDisposable
 
     int ExecuteTransaction(IList<IUpdateEntry> entries, IDiagnosticsLogger<DbLoggerCategory.Update> updateLogger);
 
+    string ClusterId { get; }
+
     KafkaOptionsExtension Options { get; }
 }

src/net/KEFCore/Storage/Internal/IKafkaClusterCache.cs

@@ -23,4 +23,6 @@ namespace MASES.EntityFrameworkCore.KNet.Storage.Internal;
 public interface IKafkaClusterCache
 {
     IKafkaCluster GetCluster(KafkaOptionsExtension options);
+
+    void Dispose(IKafkaCluster cluster);
 }

src/net/KEFCore/Storage/Internal/IKafkaTableFactory.cs

@@ -21,4 +21,6 @@ namespace MASES.EntityFrameworkCore.KNet.Storage.Internal;
 public interface IKafkaTableFactory
 {
     IKafkaTable Create(IKafkaCluster cluster, IEntityType entityType);
+
+    void Dispose(IKafkaTable table);
 }