Modernize to .NET 10, fix latent thread synchronization related Race conditions

.gitignore

@@ -26,6 +26,7 @@ bld/
 ## See more on package restore for Nuget 2.7 here: http://docs.nuget.org/consume/package-restore
 # Ignore NuGet Packages
 *.nupkg
+nupkg/
 # Ignore the packages folder
 **/packages/*
 # except build/, which is used as an MSBuild target.
@@ -58,3 +59,5 @@ $tf/pendingchanges.tfb
 $tf/properties.tf1
 project.lock.json
 **/.vs/
+
+.claude/

PACKAGE_README.md

@@ -0,0 +1,20 @@
+# Trill.StreamProcessing
+
+Trill is a high-performance one-pass in-memory streaming analytics engine originally developed by Microsoft Research. It can handle both real-time and offline data, and is based on a temporal data and query model. Trill can be used as a streaming engine, a lightweight in-memory relational engine, and as a progressive query processor for early query results on partial data.
+
+This package is a community-maintained fork of [Microsoft's Trill](https://github.com/microsoft/Trill), updated for .NET 10 and actively maintained by [Amir Burbea](https://github.com/amirburbea).
+
+## Installation
+
+```
+dotnet add package Trill.StreamProcessing
+```
+
+This package contains `Microsoft.StreamProcessing` (the core streaming engine). The experimental `Microsoft.StreamProcessing.Provider` project in the repo is not published as part of this package.
+
+## Learn More
+
+- [Trill on GitHub](https://github.com/amirburbea/trill)
+- [Original announcement blog post](https://azure.microsoft.com/en-us/blog/microsoft-open-sources-trill-to-deliver-insights-on-a-trillion-events-a-day/)
+- [Trill paper (VLDB 2015)](https://www.microsoft.com/en-us/research/wp-content/uploads/2016/02/trill-vldb2015.pdf)
+- [Trill samples repository](https://github.com/Microsoft/TrillSamples)

Sources/.editorconfig

@@ -50,4 +50,4 @@ csharp_new_line_before_catch = true
 csharp_new_line_before_finally = true
 csharp_indent_case_contents = true
 csharp_indent_switch_labels = true
-csharp_preserve_single_line_statements = false
+csharp_preserve_single_line_statements = false

Sources/Core/Directory.Build.props

@@ -3,15 +3,5 @@
   <Import Project="$([MSBuild]::GetPathOfFileAbove('Directory.Build.props', '$(MSBuildThisFileDirectory)../'))" />
 
   <!-- Code Analysis -->
-  <ItemGroup Condition="'$(TargetFramework)' == 'netstandard2.0' And '$(Configuration)' == 'Debug'">
-    <PackageReference Include="StyleCop.Analyzers" Version="1.0.2">
-      <PrivateAssets>all</PrivateAssets>
-      <IncludeAssets>runtime; build; native; contentfiles; analyzers</IncludeAssets>
-    </PackageReference>
-  </ItemGroup>
-  <PropertyGroup Condition="'$(TargetFramework)' == 'netstandard2.0'">
-    <!-- Do not set RunCodeAnalysis*, as this is incompatible with netstandard2.0. Analyzers run by default whenever included. -->
-    <CodeAnalysisRuleSet>$(EnlistmentRoot)\Sources\Microsoft.StreamProcessing.ruleset</CodeAnalysisRuleSet>
-  </PropertyGroup>
 
 </Project>

Sources/Core/Microsoft.StreamProcessing.Provider/Microsoft.StreamProcessing.Provider.csproj

@@ -1,29 +1,18 @@
 <Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
-    <TargetFrameworks>netstandard2.0</TargetFrameworks>
+    <TargetFrameworks>net10.0</TargetFrameworks>
     <Platforms>x64;AnyCPU</Platforms>
   </PropertyGroup>
 
   <PropertyGroup>
     <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
     <RootNamespace>Microsoft.StreamProcessing.Provider</RootNamespace>
+    <IsPackable>false</IsPackable>
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Microsoft.CSharp" Version="4.5.0" />
-  </ItemGroup>
-
-  <ItemGroup>
-    <PackageReference Include="Microsoft.CodeAnalysis.Scripting" Version="3.1.0" />
-    <PackageReference Include="System.Diagnostics.Contracts" Version="4.3.0" />
-    <PackageReference Include="System.Diagnostics.Process" Version="4.3.0" />
-    <PackageReference Include="System.Runtime.Serialization.Primitives" Version="4.3.0" />
-    <PackageReference Include="System.Threading.Thread" Version="4.3.0" />
-  </ItemGroup>
-
-  <ItemGroup Condition="'$(TargetFramework)'=='netstandard2.0'">
-    <PackageReference Include="System.Runtime.Loader" Version="4.3.0" />
+    <PackageReference Include="Microsoft.CodeAnalysis.Scripting" Version="5.3.0" />
   </ItemGroup>
 
   <ItemGroup>

Sources/Core/Microsoft.StreamProcessing/Aggregates/AggregateFunctions.cs

@@ -48,8 +48,8 @@ public static IAggregate<TInput, TState, TResult> Wrap<TInput, TAggregateInput,
             this IAggregate<TAggregateInput, TState, TResult> aggregate,
             Expression<Func<TInput, TAggregateInput>> transform)
         {
-            Contract.Requires(aggregate != null);
-            Contract.Requires(transform != null);
+            ArgumentNullException.ThrowIfNull(aggregate);
+            ArgumentNullException.ThrowIfNull(transform);
             return aggregate.TransformInput(transform);
         }
 
@@ -62,17 +62,17 @@ public static IAggregate<TInput, TState, TResult> Wrap<TInput, TAggregateInput,
             this IAggregate<TAggregateInput, TState, TResult> aggregate,
             Expression<Func<TInput, TAggregateInput?>> transform) where TAggregateInput : struct
         {
-            Contract.Requires(aggregate != null);
-            Contract.Requires(transform != null);
+            ArgumentNullException.ThrowIfNull(aggregate);
+            ArgumentNullException.ThrowIfNull(transform);
             return aggregate.MakeInputNullableAndSkipNulls().TransformInput(transform);
         }
 
         internal static IAggregate<TInput, TState, TResult> TransformInput<TInput, TAggregateInput, TState, TResult>(
             this IAggregate<TAggregateInput, TState, TResult> aggregate,
             Expression<Func<TInput, TAggregateInput>> transform)
         {
-            Contract.Requires(aggregate != null);
-            Contract.Requires(transform != null);
+            ArgumentNullException.ThrowIfNull(aggregate);
+            ArgumentNullException.ThrowIfNull(transform);
 
             return GeneratedAggregate.Create(
                 initialState: aggregate.InitialState(),
@@ -86,19 +86,19 @@ private static Expression<Func<T1, T2, TInput, TOutput>> TransformInput3<T1, T2,
             this Expression<Func<T1, T2, T3, TOutput>> func,
             Expression<Func<TInput, T3>> transform)
         {
-            Contract.Requires(func != null);
-            Contract.Requires(transform != null);
+            ArgumentNullException.ThrowIfNull(func);
+            ArgumentNullException.ThrowIfNull(transform);
             var result = func.ReplaceParametersInBody(func.Parameters[0], func.Parameters[1], transform.Body);
             var transformParam = transform.Parameters[0];
-            return Expression.Lambda<Func<T1, T2, TInput, TOutput>>(result, new[] { func.Parameters[0], func.Parameters[1], transformParam });
+            return Expression.Lambda<Func<T1, T2, TInput, TOutput>>(result, [func.Parameters[0], func.Parameters[1], transformParam]);
         }
 
         internal static IAggregate<TInput, TState, TResult> TransformOutput<TInput, TState, TAggregateResult, TResult>(
             this IAggregate<TInput, TState, TAggregateResult> aggregate,
             Expression<Func<TAggregateResult, TResult>> transform)
         {
-            Contract.Requires(aggregate != null);
-            Contract.Requires(transform != null);
+            ArgumentNullException.ThrowIfNull(aggregate);
+            ArgumentNullException.ThrowIfNull(transform);
 
             return GeneratedAggregate.Create(
                 initialState: aggregate.InitialState(),
@@ -112,8 +112,8 @@ private static Expression<Func<T1, TOutput>> TransformOutput<T1, TFuncOutput, TO
             this Expression<Func<T1, TFuncOutput>> func,
             Expression<Func<TFuncOutput, TOutput>> transform)
         {
-            Contract.Requires(func != null);
-            Contract.Requires(transform != null);
+            ArgumentNullException.ThrowIfNull(func);
+            ArgumentNullException.ThrowIfNull(transform);
             var result = transform.ReplaceParametersInBody(func.Body);
             return Expression.Lambda<Func<T1, TOutput>>(result, func.Parameters);
         }
@@ -127,7 +127,7 @@ public static IAggregate<TInput, TState, TResult> ApplyFilter<TInput, TState, TR
             this IAggregate<TInput, TState, TResult> aggregate,
             Expression<Func<TInput, bool>> filter)
         {
-            Contract.Requires(aggregate != null);
+            ArgumentNullException.ThrowIfNull(aggregate);
             if (filter == null || filter.Body.ExpressionEquals(Expression.Constant(true))) return aggregate;
 
             Expression<Func<TState, long, TInput, TState>> newAccumulate = (oldState, timestamp, input) =>
@@ -151,7 +151,7 @@ public static IAggregate<TInput, TState, TResult> ApplyFilter<TInput, TState, TR
         public static IAggregate<TInput?, TState, TResult> MakeInputNullableAndSkipNulls<TInput, TState, TResult>(
             this IAggregate<TInput, TState, TResult> aggregate) where TInput : struct
         {
-            Contract.Requires(aggregate != null);
+            ArgumentNullException.ThrowIfNull(aggregate);
 
             Expression<Func<TState, long, TInput?, TState>> newAccumulate = (oldState, timestamp, input) =>
                 input.HasValue ? CallInliner.Call(aggregate.Accumulate(), oldState, timestamp, input.Value) : oldState;
@@ -175,9 +175,9 @@ public static IAggregate<TInput, TState, TResult> ApplyFilter<TInput, TState, TR
         public static IAggregate<TInput, TState, TResult> SkipNulls<TInput, TState, TResult>(
             this IAggregate<TInput, TState, TResult> aggregate)
         {
-            Contract.Requires(aggregate != null);
+            ArgumentNullException.ThrowIfNull(aggregate);
 
-            var inputType = typeof(TInput).GetTypeInfo();
+            var inputType = typeof(TInput);
             return inputType.IsClass
                 ? GeneratedAggregate.Create(
                     initialState: aggregate.InitialState(),
@@ -229,7 +229,7 @@ private static Expression<Func<TState, long, TInput, TState>> AddSkipNullValueLo
         public static IAggregate<TInput, NullOutputWrapper<TState>, TResult?> MakeOutputNullableAndOutputNullWhenEmpty<TInput, TState, TResult>(
             this IAggregate<TInput, TState, TResult> aggregate) where TResult : struct
         {
-            Contract.Requires(aggregate != null);
+            ArgumentNullException.ThrowIfNull(aggregate);
 
             Expression<Func<NullOutputWrapper<TState>>> newInitialState =
                 () => new NullOutputWrapper<TState>
@@ -279,7 +279,7 @@ private static Expression<Func<TState, long, TInput, TState>> AddSkipNullValueLo
         public static IAggregate<TInput, NullOutputWrapper<TState>, TResult> OutputDefaultWhenEmpty<TInput, TState, TResult>(
             this IAggregate<TInput, TState, TResult> aggregate)
         {
-            Contract.Requires(aggregate != null);
+            ArgumentNullException.ThrowIfNull(aggregate);
 
             Expression<Func<NullOutputWrapper<TState>>> newInitialState =
                 () => new NullOutputWrapper<TState>