Merge pull request #3447 from akkadotnet/dev

v1.3.7 Production Akka.NET Release
akkadotnet · May 15, 2018 · bb80ab3 · bb80ab3
2 parents dc7c581 + dbb2081
commit bb80ab3
Show file tree

Hide file tree

Showing 21 changed files with 209 additions and 108 deletions.
diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md
@@ -1,3 +1,32 @@
+#### 1.3.7 April 17 2018 ####
+**Maintenance Release for Akka.NET 1.3**
+
+Akka.NET v1.3.7 is a minor patch consisting mostly of bug fixes.
+
+**DotNetty stabilization**
+We've had a number of issues related to DotNetty issues over recent weeks, and we've resolved those in this patch by doing the following:
+
+* [Locking down the version of DotNetty to v0.4.6 until further notice](https://github.com/akkadotnet/akka.net/pull/3410)
+* [Resolving memory leaks introduced with DotNetty in v1.3.6](https://github.com/akkadotnet/akka.net/pull/3436)
+
+We will be upgrading to DotNetty v0.4.8 in a near future release, but in the meantime this patch fixes critical issues introduced in v1.3.6.
+
+**Bugfixes**
+1. [Akka.Persistence.Sql: Slow reading of big snapshots](https://github.com/akkadotnet/akka.net/issues/3422) - this will require a recompilation of all Akka.Persistence.Sql-type Akka.Persistence plugins.
+2. [Akka.Fsharp: spawning an actor results in Exception in 1.3.6 release](https://github.com/akkadotnet/akka.net/issues/3402)
+
+See [the full list of fixes for Akka.NET v1.3.7 here](https://github.com/akkadotnet/akka.net/milestone/25).
+
+| COMMITS | LOC+ | LOC- | AUTHOR |
+| --- | --- | --- | --- |
+| 5 | 130 | 180 | Aaron Stannard |
+| 3 | 7 | 1 | chrisjhoare |
+| 2 | 3 | 1 | ivog |
+| 1 | 70 | 17 | TietoOliverKurowski |
+| 1 | 41 | 4 | Bart de Boer |
+| 1 | 11 | 3 | Oleksandr Bogomaz |
+| 1 | 1 | 1 | Vasily Kirichenko |
+
 #### 1.3.6 April 17 2018 ####
 **Maintenance Release for Akka.NET 1.3**
 

diff --git a/build.fsx b/build.fsx
@@ -282,9 +282,17 @@ Target "NBench" <| fun _ ->
             info.FileName <- nbenchTestPath
             info.WorkingDirectory <- (Path.GetDirectoryName (FullName nbenchTestPath))
             info.Arguments <- args) (System.TimeSpan.FromMinutes 45.0) (* Reasonably long-running task. *)
-        if result <> 0 then failwithf "NBench.Runner failed. %s %s" nbenchTestPath args
-
-    nbenchTestAssemblies |> Seq.iter runNBench
+        if result <> 0 then failwithf "%s %s \nexited with code %i" nbenchTestPath args result
+
+    let failedRuns =
+        nbenchTestAssemblies
+        |> Seq.map (fun asm -> try runNBench asm; None with e -> Some(e.ToString()))
+        |> Seq.filter Option.isSome
+        |> Seq.map Option.get
+        |> Seq.mapi (fun i s -> sprintf "%i: \"%s\"" (i + 1) s)
+        |> Seq.toArray
+    if failedRuns.Length > 0 then
+        failwithf "NBench.Runner failed for %i run(s):\n%s\n\n" failedRuns.Length (String.concat "\n\n" failedRuns)
 
 //--------------------------------------------------------------------------------
 // Nuget targets 

diff --git a/buildIncremental.fsx b/buildIncremental.fsx
@@ -42,6 +42,7 @@ module IncrementalTests =
 
     let getUnitTestProjects runtime =
         let allTestProjects = !! "./**/core/**/*.Tests.csproj"
+                              ++ "./**/core/**/*.Tests.fsproj"
                               ++ "./**/contrib/**/*.Tests.csproj"
                               -- "./**/serializers/**/*Wire*.csproj"
         allTestProjects 

diff --git a/src/common.props b/src/common.props
@@ -2,7 +2,7 @@
   <PropertyGroup>
     <Copyright>Copyright © 2013-2017 Akka.NET Team</Copyright>
     <Authors>Akka.NET Team</Authors>
-    <VersionPrefix>1.3.6</VersionPrefix>
+    <VersionPrefix>1.3.7</VersionPrefix>
     <PackageIconUrl>http://getakka.net/images/akkalogo.png</PackageIconUrl>
     <PackageProjectUrl>https://github.com/akkadotnet/akka.net</PackageProjectUrl>
     <PackageLicenseUrl>https://github.com/akkadotnet/akka.net/blob/master/LICENSE</PackageLicenseUrl>
@@ -18,37 +18,24 @@
   </PropertyGroup>
   <PropertyGroup>
     <PackageReleaseNotes>Maintenance Release for Akka.NET 1.3**
-Akka.NET v1.3.6 is a minor patch consisting mostly of bug fixes.
-Akka.FSharp on .NET Standard**
-The biggest change in this release is [the availability of Akka.FSharp on .NET Standard and .NET Core](https://github.com/akkadotnet/akka.net/issues/2826)!
-Akka.FSharp runs on .NET Standard 2.0 as of 1.3.6 (it doesn't support .NET Standard 1.6 like the rest of Akka.NET due to FSharp-specific, downstream dependencies.)
-Updates and Additions**
-1. [Akka.Streams: Port 4 "streams contrib" stages - AccumulateWhileUnchanged, LastElement, PartitionWith, Sample](https://github.com/akkadotnet/akka.net/pull/3375)
-2. [Akka.Remote: Add `public-port` setting to allow for port aliasing inside environments like Docker, PCF](https://github.com/akkadotnet/akka.net/issues/3357)
+Akka.NET v1.3.7 is a minor patch consisting mostly of bug fixes.
+DotNetty stabilization**
+We've had a number of issues related to DotNetty issues over recent weeks, and we've resolved those in this patch by doing the following:
+[Locking down the version of DotNetty to v0.4.6 until further notice](https://github.com/akkadotnet/akka.net/pull/3410)
+[Resolving memory leaks introduced with DotNetty in v1.3.6](https://github.com/akkadotnet/akka.net/pull/3436)
+We will be upgrading to DotNetty v0.4.8 in a near future release, but in the meantime this patch fixes critical issues introduced in v1.3.6.
 Bugfixes**
-1. [Akka.Cluster.Sharding: Removing string.GetHashCode usage from distributed classes](https://github.com/akkadotnet/akka.net/pull/3363)
-2. [Akka.Cluster.Sharding: HashCodeMessageExtractor can create inconsistent ShardId's](https://github.com/akkadotnet/akka.net/issues/3361)
-3. [Akka.Remote:
-Error while decoding incoming Akka PDU Exception when communicating between Remote Actors with a large number of messages on Linux](https://github.com/akkadotnet/akka.net/issues/3370)
-4. [Akka.Cluster.Sharding: DData: Cannot create a shard proxy on a cluster node that is not in the same role as the proxied shard entity](https://github.com/akkadotnet/akka.net/issues/3352)
-5. [Akka.Streams: Fix GroupedWithin allocation of new buffer after emit.](https://github.com/akkadotnet/akka.net/pull/3382)
-6. [Akka.Persistence: Add missing ReturnRecoveryPermit](https://github.com/akkadotnet/akka.net/pull/3372)
-You can see [the full set of changes for Akka.NET v1.3.6 here](hhttps://github.com/akkadotnet/akka.net/milestone/24).
+1. [Akka.Persistence.Sql: Slow reading of big snapshots](https://github.com/akkadotnet/akka.net/issues/3422) - this will require a recompilation of all Akka.Persistence.Sql-type Akka.Persistence plugins.
+2. [Akka.Fsharp: spawning an actor results in Exception in 1.3.6 release](https://github.com/akkadotnet/akka.net/issues/3402)
+See [the full list of fixes for Akka.NET v1.3.7 here](https://github.com/akkadotnet/akka.net/milestone/25).
 | COMMITS | LOC+ | LOC- | AUTHOR |
 | --- | --- | --- | --- |
-| 7 | 261 | 38 | Aaron Stannard |
-| 6 | 28 | 28 | cimryan |
-| 5 | 53 | 20 | Tomasz Jaskula |
-| 2 | 7 | 4 | Ondrej Pialek |
-| 2 | 20 | 10 | Ismael Hamed |
-| 1 | 739 | 0 | Oleksandr Bogomaz |
-| 1 | 64 | 6 | Robert |
-| 1 | 23 | 29 | nathvi |
-| 1 | 2 | 1 | Sebastien Bacquet |
-| 1 | 1 | 2 | OndÅej PiÃ¡lek |
-| 1 | 1 | 1 | Steffen Skov |
-| 1 | 1 | 1 | Sean Gilliam |
-| 1 | 1 | 1 | Matthew Herman |
-| 1 | 1 | 1 | Jan Pluskal |</PackageReleaseNotes>
+| 5 | 130 | 180 | Aaron Stannard |
+| 3 | 7 | 1 | chrisjhoare |
+| 2 | 3 | 1 | ivog |
+| 1 | 70 | 17 | TietoOliverKurowski |
+| 1 | 41 | 4 | Bart de Boer |
+| 1 | 11 | 3 | Oleksandr Bogomaz |
+| 1 | 1 | 1 | Vasily Kirichenko |</PackageReleaseNotes>
   </PropertyGroup>
 </Project>
diff --git a/src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/QueryExecutor.cs b/src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/QueryExecutor.cs
@@ -38,8 +38,8 @@ public interface IJournalQueryExecutor
         Task<ImmutableArray<string>> SelectAllPersistenceIdsAsync(DbConnection connection, CancellationToken cancellationToken);
 
         /// <summary>
-        /// Asynchronously replays a <paramref name="callback"/> on all selected events for provided 
-        /// <paramref name="persistenceId"/>, within boundaries of <paramref name="fromSequenceNr"/> 
+        /// Asynchronously replays a <paramref name="callback"/> on all selected events for provided
+        /// <paramref name="persistenceId"/>, within boundaries of <paramref name="fromSequenceNr"/>
         /// and <paramref name="toSequenceNr"/> up to <paramref name="max"/> number of events.
         /// </summary>
         /// <param name="connection">TBD</param>
@@ -53,8 +53,8 @@ public interface IJournalQueryExecutor
         Task SelectByPersistenceIdAsync(DbConnection connection, CancellationToken cancellationToken, string persistenceId, long fromSequenceNr, long toSequenceNr, long max, Action<IPersistentRepresentation> callback);
 
         /// <summary>
-        /// Asynchronously replays <paramref name="callback"/> on all selected events, which have been tagged using 
-        /// provided <paramref name="tag"/>, within boundaries of <paramref name="fromOffset"/> and 
+        /// Asynchronously replays <paramref name="callback"/> on all selected events, which have been tagged using
+        /// provided <paramref name="tag"/>, within boundaries of <paramref name="fromOffset"/> and
         /// <paramref name="toOffset"/>, up to <paramref name="max"/> number of elements.
         /// Returns highest sequence number from selected events.
         /// </summary>
@@ -171,6 +171,11 @@ public class QueryConfiguration
         /// </summary>
         public string DefaultSerializer { get; }
 
+        /// <summary>
+        /// Uses the CommandBehavior.SequentialAccess when creating the command, providing a performance improvement for reading large BLOBS.
+        /// </summary>
+        public bool UseSequentialAccess { get; }
+
         /// <summary>
         /// TBD
         /// </summary>
@@ -188,6 +193,7 @@ public class QueryConfiguration
         /// <param name="serializerIdColumnName">TBD</param>
         /// <param name="timeout">TBD</param>
         /// <param name="defaultSerializer">The default serializer used when not type override matching is found</param>
+        /// <param name="useSequentialAccess">Uses the CommandBehavior.SequentialAccess when creating the command, providing a performance improvement for reading large BLOBS.</param>
         public QueryConfiguration(
             string schemaName,
             string journalEventsTableName,
@@ -202,7 +208,8 @@ public QueryConfiguration(
             string orderingColumnName,
             string serializerIdColumnName,
             TimeSpan timeout,
-            string defaultSerializer)
+            string defaultSerializer,
+            bool useSequentialAccess)
         {
             SchemaName = schemaName;
             JournalEventsTableName = journalEventsTableName;
@@ -218,6 +225,7 @@ public QueryConfiguration(
             OrderingColumnName = orderingColumnName;
             DefaultSerializer = defaultSerializer;
             SerializerIdColumnName = serializerIdColumnName;
+            UseSequentialAccess = useSequentialAccess;
         }
 
         /// <summary>
@@ -235,7 +243,7 @@ public QueryConfiguration(
     /// </summary>
     public abstract class AbstractQueryExecutor : IJournalQueryExecutor
     {
-        // indexes of particular fields returned from all events queries 
+        // indexes of particular fields returned from all events queries
         // they must match `allEventColumnNames` order
         /// <summary>
         /// TBD
@@ -461,7 +469,18 @@ public virtual async Task SelectByPersistenceIdAsync(DbConnection connection, Ca
                 AddParameter(command, "@FromSequenceNr", DbType.Int64, fromSequenceNr);
                 AddParameter(command, "@ToSequenceNr", DbType.Int64, toSequenceNr);
 
-                using (var reader = await command.ExecuteReaderAsync(cancellationToken))
+                CommandBehavior commandBehavior;
+
+                if (Configuration.UseSequentialAccess)
+                {
+                    commandBehavior = CommandBehavior.SequentialAccess;
+                }
+                else
+                {
+                    commandBehavior = CommandBehavior.Default;
+                }
+
+                using (var reader = await command.ExecuteReaderAsync(commandBehavior, cancellationToken))
                 {
                     var i = 0L;
                     while ((i++) < max && await reader.ReadAsync(cancellationToken))
@@ -494,7 +513,18 @@ public virtual async Task<long> SelectByTagAsync(DbConnection connection, Cancel
                 AddParameter(command, "@Ordering", DbType.Int64, fromOffset);
                 AddParameter(command, "@Take", DbType.Int64, take);
 
-                using (var reader = await command.ExecuteReaderAsync(cancellationToken))
+                CommandBehavior commandBehavior;
+
+                if (Configuration.UseSequentialAccess)
+                {
+                    commandBehavior = CommandBehavior.SequentialAccess;
+                }
+                else
+                {
+                    commandBehavior = CommandBehavior.Default;
+                }
+
+                using (var reader = await command.ExecuteReaderAsync(commandBehavior, cancellationToken))
                 {
                     var maxSequenceNr = 0L;
                     while (await reader.ReadAsync(cancellationToken))

diff --git a/src/contrib/persistence/Akka.Persistence.Sql.Common/Snapshot/QueryExecutor.cs b/src/contrib/persistence/Akka.Persistence.Sql.Common/Snapshot/QueryExecutor.cs
@@ -16,7 +16,7 @@
 namespace Akka.Persistence.Sql.Common.Snapshot
 {
     /// <summary>
-    /// Flattened and serialized snapshot object used as intermediate representation 
+    /// Flattened and serialized snapshot object used as intermediate representation
     /// before saving snapshot with metadata inside SQL Server database.
     /// </summary>
     public class SnapshotEntry
@@ -112,6 +112,11 @@ public class QueryConfiguration
         /// </summary>
         public readonly string DefaultSerializer;
 
+        /// <summary>
+        /// Uses the CommandBehavior.SequentialAccess when creating the command, providing a performance improvement for reading large BLOBS.
+        /// </summary>
+        public bool UseSequentialAccess { get; }
+
         /// <summary>
         /// TBD
         /// </summary>
@@ -125,6 +130,7 @@ public class QueryConfiguration
         /// <param name="serializerIdColumnName">TBD</param>
         /// <param name="timeout">TBD</param>
         /// <param name="defaultSerializer">The default serializer used when not type override matching is found</param>
+        /// <param name="useSequentialAccess">Uses the CommandBehavior.SequentialAccess when creating the command, providing a performance improvement for reading large BLOBS.</param>
         public QueryConfiguration(
             string schemaName,
             string snapshotTableName,
@@ -134,8 +140,9 @@ public QueryConfiguration(
             string manifestColumnName,
             string timestampColumnName,
             string serializerIdColumnName,
-            TimeSpan timeout, 
-            string defaultSerializer)
+            TimeSpan timeout,
+            string defaultSerializer,
+            bool useSequentialAccess)
         {
             SchemaName = schemaName;
             SnapshotTableName = snapshotTableName;
@@ -147,6 +154,7 @@ public QueryConfiguration(
             SerializerIdColumnName = serializerIdColumnName;
             Timeout = timeout;
             DefaultSerializer = defaultSerializer;
+            UseSequentialAccess = useSequentialAccess;
         }
 
         /// <summary>
@@ -166,7 +174,7 @@ public interface ISnapshotQueryExecutor
         QueryConfiguration Configuration { get; }
 
         /// <summary>
-        /// Deletes a single snapshot identified by it's persistent actor's <paramref name="persistenceId"/>, 
+        /// Deletes a single snapshot identified by it's persistent actor's <paramref name="persistenceId"/>,
         /// <paramref name="sequenceNr"/> and <paramref name="timestamp"/>.
         /// </summary>
         /// <param name="connection">TBD</param>
@@ -178,7 +186,7 @@ public interface ISnapshotQueryExecutor
         Task DeleteAsync(DbConnection connection, CancellationToken cancellationToken, string persistenceId, long sequenceNr, DateTime? timestamp);
 
         /// <summary>
-        /// Deletes all snapshot matching persistent actor's <paramref name="persistenceId"/> as well as 
+        /// Deletes all snapshot matching persistent actor's <paramref name="persistenceId"/> as well as
         /// upper (inclusive) bounds of the both <paramref name="maxSequenceNr"/> and <paramref name="maxTimestamp"/>.
         /// </summary>
         /// <param name="connection">TBD</param>
@@ -466,7 +474,19 @@ public virtual async Task<SelectedSnapshot> SelectSnapshotAsync(DbConnection con
                 SetPersistenceIdParameter(persistenceId, command);
                 SetSequenceNrParameter(maxSequenceNr, command);
                 SetTimestampParameter(maxTimestamp, command);
-                using (var reader = await command.ExecuteReaderAsync(cancellationToken))
+
+                CommandBehavior commandBehavior;
+
+                if (Configuration.UseSequentialAccess)
+                {
+                    commandBehavior = CommandBehavior.SequentialAccess;
+                }
+                else
+                {
+                    commandBehavior = CommandBehavior.Default;
+                }
+
+                using (var reader = await command.ExecuteReaderAsync(commandBehavior, cancellationToken))
                 {
                     if (await reader.ReadAsync(cancellationToken))
                     {

diff --git a/src/contrib/persistence/Akka.Persistence.Sqlite/Journal/BatchingSqliteJournal.cs b/src/contrib/persistence/Akka.Persistence.Sqlite/Journal/BatchingSqliteJournal.cs
@@ -40,7 +40,8 @@ public sealed class BatchingSqliteJournalSetup : BatchingSqlJournalSetup
                     orderingColumnName: "ordering",
                     serializerIdColumnName: "serializer_id",
                     timeout: config.GetTimeSpan("connection-timeout"),
-                    defaultSerializer: config.GetString("serializer")))
+                    defaultSerializer: config.GetString("serializer"),
+                    useSequentialAccess: config.GetBoolean("use-sequential-access")))
         {
         }
 

diff --git a/src/contrib/persistence/Akka.Persistence.Sqlite/Journal/SqliteJournal.cs b/src/contrib/persistence/Akka.Persistence.Sqlite/Journal/SqliteJournal.cs
@@ -43,7 +43,8 @@ public SqliteJournal(Config journalConfig) : base(journalConfig.WithFallback(Ext
                 orderingColumnName: "ordering",
                 serializerIdColumnName: "serializer_id",
                 timeout: config.GetTimeSpan("connection-timeout"),
-                defaultSerializer: config.GetString("serializer")), 
+                defaultSerializer: config.GetString("serializer"),
+                useSequentialAccess: config.GetBoolean("use-sequential-access")), 
                     Context.System.Serialization, 
                     GetTimestampProvider(config.GetString("timestamp-provider")));
         }

diff --git a/src/contrib/persistence/Akka.Persistence.Sqlite/Snapshot/SqliteSnapshotStore.cs b/src/contrib/persistence/Akka.Persistence.Sqlite/Snapshot/SqliteSnapshotStore.cs
@@ -121,7 +121,8 @@ public SqliteSnapshotStore(Config snapshotConfig) : base(snapshotConfig)
                 timestampColumnName: "created_at",
                 serializerIdColumnName: "serializer_id",
                 timeout: config.GetTimeSpan("connection-timeout"),
-                defaultSerializer: config.GetString("serializer")), 
+                defaultSerializer: config.GetString("serializer"),
+                useSequentialAccess: config.GetBoolean("use-sequential-access")),
                 Context.System.Serialization);
         }
 

diff --git a/src/core/Akka.API.Tests/CoreAPISpec.ApproveCore.approved.txt b/src/core/Akka.API.Tests/CoreAPISpec.ApproveCore.approved.txt
@@ -883,7 +883,8 @@ namespace Akka.Actor
         public static System.Threading.Tasks.Task<object> Ask(this Akka.Actor.ICanTell self, object message, System.Nullable<System.TimeSpan> timeout, System.Threading.CancellationToken cancellationToken) { }
         public static System.Threading.Tasks.Task<T> Ask<T>(this Akka.Actor.ICanTell self, object message, System.Nullable<System.TimeSpan> timeout = null) { }
         public static System.Threading.Tasks.Task<T> Ask<T>(this Akka.Actor.ICanTell self, object message, System.Threading.CancellationToken cancellationToken) { }
-        public static async System.Threading.Tasks.Task<T> Ask<T>(this Akka.Actor.ICanTell self, object message, System.Nullable<System.TimeSpan> timeout, System.Threading.CancellationToken cancellationToken) { }
+        public static System.Threading.Tasks.Task<T> Ask<T>(this Akka.Actor.ICanTell self, object message, System.Nullable<System.TimeSpan> timeout, System.Threading.CancellationToken cancellationToken) { }
+        public static async System.Threading.Tasks.Task<T> Ask<T>(this Akka.Actor.ICanTell self, System.Func<Akka.Actor.IActorRef, object> messageFactory, System.Nullable<System.TimeSpan> timeout, System.Threading.CancellationToken cancellationToken) { }
     }
     public class static GracefulStopSupport
     {