Fix persistence allocation. (#6487)

Co-authored-by: Aaron Stannard <aaron@petabridge.com>
akkadotnet · Mar 6, 2023 · 149cc05 · 149cc05
1 parent bfe6636
commit 149cc05
Show file tree

Hide file tree

Showing 4 changed files with 30 additions and 78 deletions.
diff --git a/src/core/Akka.Persistence/Eventsourced.Recovery.cs b/src/core/Akka.Persistence/Eventsourced.Recovery.cs
@@ -6,8 +6,6 @@
 //-----------------------------------------------------------------------
 
 using System;
-using System.Collections.Generic;
-using System.Linq;
 using Akka.Actor;
 using Akka.Persistence.Internal;
 
@@ -334,11 +332,11 @@ private void FlushBatch()
         {
             if (_eventBatch.Count > 0)
             {
-                foreach (var p in _eventBatch.Reverse())
+                foreach (var p in _eventBatch)
                 {
                     _journalBatch.Add(p);
                 }
-                _eventBatch = new LinkedList<IPersistentEnvelope>();
+                _eventBatch.Clear();
             }
 
             FlushJournalBatch();

diff --git a/src/core/Akka.Persistence/Eventsourced.cs b/src/core/Akka.Persistence/Eventsourced.cs
@@ -303,7 +303,7 @@ public void Persist<TEvent>(TEvent @event, Action<TEvent> handler)
 
             _pendingStashingPersistInvocations++;
             _pendingInvocations.AddLast(new StashingHandlerInvocation(@event, o => handler((TEvent)o)));
-            _eventBatch.AddFirst(new AtomicWrite(new Persistent(@event, persistenceId: PersistenceId,
+            _eventBatch.AddLast(new AtomicWrite(new Persistent(@event, persistenceId: PersistenceId,
                 sequenceNr: NextSequenceNr(), writerGuid: _writerGuid, sender: Sender)));
         }
 
@@ -335,7 +335,7 @@ public void PersistAll<TEvent>(IEnumerable<TEvent> events, Action<TEvent> handle
             }
 
             if (persistents.Count > 0)
-                _eventBatch.AddFirst(new AtomicWrite(persistents.ToImmutable()));
+                _eventBatch.AddLast(new AtomicWrite(persistents.ToImmutable()));
         }
 
         /// <summary>
@@ -374,7 +374,7 @@ public void PersistAsync<TEvent>(TEvent @event, Action<TEvent> handler)
             }
 
             _pendingInvocations.AddLast(new AsyncHandlerInvocation(@event, o => handler((TEvent)o)));
-            _eventBatch.AddFirst(new AtomicWrite(new Persistent(@event, persistenceId: PersistenceId,
+            _eventBatch.AddLast(new AtomicWrite(new Persistent(@event, persistenceId: PersistenceId,
                 sequenceNr: NextSequenceNr(), writerGuid: _writerGuid, sender: Sender)));
         }
 
@@ -400,7 +400,7 @@ public void PersistAllAsync<TEvent>(IEnumerable<TEvent> events, Action<TEvent> h
                 _pendingInvocations.AddLast(new AsyncHandlerInvocation(@event, Inv));
             }
 
-            _eventBatch.AddFirst(new AtomicWrite(enumerable.Select(e => new Persistent(e, persistenceId: PersistenceId,
+            _eventBatch.AddLast(new AtomicWrite(enumerable.Select(e => new Persistent(e, persistenceId: PersistenceId,
                     sequenceNr: NextSequenceNr(), writerGuid: _writerGuid, sender: Sender))
                 .ToImmutableList<IPersistentRepresentation>()));
         }
@@ -440,7 +440,7 @@ public void DeferAsync<TEvent>(TEvent evt, Action<TEvent> handler)
             else
             {
                 _pendingInvocations.AddLast(new AsyncHandlerInvocation(evt, o => handler((TEvent)o)));
-                _eventBatch.AddFirst(new NonPersistentMessage(evt, Sender));
+                _eventBatch.AddLast(new NonPersistentMessage(evt, Sender));
             }
         }
 

diff --git a/src/core/Akka.Persistence/Journal/AsyncWriteJournal.cs b/src/core/Akka.Persistence/Journal/AsyncWriteJournal.cs
@@ -341,7 +341,7 @@ private void HandleWriteMessages(WriteMessages message)
              */
             var self = Self;
             _resequencerCounter += message.Messages.Aggregate(1, (acc, m) => acc + m.Size);
-            var atomicWriteCount = message.Messages.OfType<AtomicWrite>().Count();
+            var atomicWriteCount = message.Messages.Count(x => x is AtomicWrite);
 
             // Using an async local function instead of ContinueWith
 #pragma warning disable CS4014