使用Immutable<T>提高事件转发的性能

Author: uless

src/Ray.Core/Core/Abstractions/IConcurrentObserver.cs

@@ -6,6 +6,6 @@ namespace Ray.Core
     public interface IConcurrentObserver : IObserver
     {
         [AlwaysInterleave]
-        Task ConcurrentOnNext(byte[] bytes);
+        Task ConcurrentOnNext(Immutable<byte[]> bytes);
     }
 }

src/Ray.Core/Core/Abstractions/IObserver.cs

@@ -1,10 +1,11 @@
 using System.Threading.Tasks;
+using Orleans.Concurrency;
 
 namespace Ray.Core
 {
     public interface IObserver
     {
-        Task OnNext(byte[] bytes);
+        Task OnNext(Immutable<byte[]> bytes);
         Task<long> GetVersion();
         Task<long> GetAndSaveVersion(long compareVersion);
     }

src/Ray.Core/Core/Grains/ConcurrentObserverGrain.cs

@@ -1,13 +1,14 @@
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Threading.Tasks;
-using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
+using Orleans.Concurrency;
 using Ray.Core.Channels;
 using Ray.Core.Event;
 using Ray.Core.Exceptions;
 using Ray.Core.Serialization;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
 
 namespace Ray.Core
 {
@@ -32,9 +33,9 @@ public override Task OnDeactivateAsync()
             ConcurrentChannel.Complete();
             return base.OnDeactivateAsync();
         }
-        public async Task ConcurrentOnNext(byte[] bytes)
+        public async Task ConcurrentOnNext(Immutable<byte[]> bytes)
         {
-            var (success, transport) = EventBytesTransport.FromBytesWithNoId(bytes);
+            var (success, transport) = EventBytesTransport.FromBytesWithNoId(bytes.Value);
             if (success)
             {
                 var data = Serializer.Deserialize(TypeContainer.GetType(transport.EventType), transport.EventBytes);

src/Ray.Core/Core/Grains/ObserverGrain.cs

@@ -1,16 +1,17 @@
-using System;
-using System.Linq;
-using System.Runtime.CompilerServices;
-using System.Threading.Tasks;
-using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
 using Orleans;
+using Orleans.Concurrency;
 using Ray.Core.Configuration;
 using Ray.Core.Event;
 using Ray.Core.Exceptions;
 using Ray.Core.Serialization;
 using Ray.Core.Snapshot;
 using Ray.Core.Storage;
+using System;
+using System.Linq;
+using System.Runtime.CompilerServices;
+using System.Threading.Tasks;
 
 namespace Ray.Core
 {
@@ -76,7 +77,7 @@ protected async virtual ValueTask DependencyInjection()
                 await eventStorageTask;
             EventStorage = eventStorageTask.Result;
             //创建状态存储器
-            var followConfigTask = configureBuilder.GetObserverConfig(ServiceProvider,GrainType, GrainId);
+            var followConfigTask = configureBuilder.GetObserverConfig(ServiceProvider, GrainType, GrainId);
             if (!followConfigTask.IsCompletedSuccessfully)
                 await followConfigTask;
             var stateStorageTask = storageFactory.CreateObserverSnapshotStorage(followConfigTask.Result, GrainId);
@@ -188,9 +189,9 @@ protected virtual ValueTask InitFirstSnapshot()
             return Consts.ValueTaskDone;
         }
         #endregion
-        public Task OnNext(byte[] bytes)
+        public Task OnNext(Immutable<byte[]> bytes)
         {
-            var (success, transport) = EventBytesTransport.FromBytesWithNoId(bytes);
+            var (success, transport) = EventBytesTransport.FromBytesWithNoId(bytes.Value);
             if (success)
             {
                 var data = Serializer.Deserialize(TypeContainer.GetType(transport.EventType), transport.EventBytes);

src/Ray.Core/Core/Grains/ShadowGrain.cs

@@ -1,15 +1,16 @@
-using System;
-using System.Runtime.CompilerServices;
-using System.Threading.Tasks;
-using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
 using Orleans;
+using Orleans.Concurrency;
 using Ray.Core.Configuration;
 using Ray.Core.Event;
 using Ray.Core.Exceptions;
 using Ray.Core.Serialization;
 using Ray.Core.Snapshot;
 using Ray.Core.Storage;
+using System;
+using System.Runtime.CompilerServices;
+using System.Threading.Tasks;
 
 namespace Ray.Core
 {
@@ -174,9 +175,9 @@ protected virtual ValueTask CreateSnapshot()
             return Consts.ValueTaskDone;
         }
         #endregion
-        public Task OnNext(byte[] bytes)
+        public Task OnNext(Immutable<byte[]> bytes)
         {
-            var (success, transport) = EventBytesTransport.FromBytesWithNoId(bytes);
+            var (success, transport) = EventBytesTransport.FromBytesWithNoId(bytes.Value);
             if (success)
             {
                 var eventType = TypeContainer.GetType(transport.EventType);

src/Ray.Core/Core/Observer/ObserverUnit.cs

@@ -1,13 +1,14 @@
 using Microsoft.Extensions.DependencyInjection;
 using Orleans;
+using Orleans.Concurrency;
 using Ray.Core.Event;
 using Ray.Core.Serialization;
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
 using System.Linq.Expressions;
 using System.Threading.Tasks;
-using System.Collections.Concurrent;
 
 namespace Ray.Core
 {
@@ -91,8 +92,8 @@ Task func(byte[] bytes)
                 {
                     var observer = GetObserver(observerType, actorId);
                     if (observer is IConcurrentObserver concurrentObserver)
-                        return concurrentObserver.ConcurrentOnNext(bytes);
-                    return observer.OnNext(bytes);
+                        return concurrentObserver.ConcurrentOnNext(new Immutable<byte[]>(bytes));
+                    return observer.OnNext(new Immutable<byte[]>(bytes));
                 }
                 return Task.CompletedTask;
             }