Parallelize BatchScheduler

.github/workflows/dot_net_format.yml

@@ -4,15 +4,15 @@ jobs:
   dotnet-format:
     runs-on: ubuntu-latest
 
-    steps:     
+    steps:
       - name: Azure login
         uses: azure/login@v1
         with:
           creds: '${{ secrets.AZURE_CREDENTIALS }}'
-          
+
       - name: Check out code
         uses: actions/checkout@v3.3.0
-        
+
       - name: Remove nuget.config
         uses: JesseTG/rm@v1.0.3
         with:

format/Setup Pre-commit.md

@@ -1,12 +1,12 @@
 # Set up Pre-commit
-To set up dotnet format to run locally on 'git commit', you need to perform the following actions. You should only need to do this once unless you remove and clone the repo again. 
+To set up dotnet format to run locally on 'git commit', you need to perform the following actions. You should only need to do this once unless you remove and clone the repo again.
 
 ## Git pre-commit hook to reformat
 Navigate to GIT-REPO-LOCATION/format. Run the "dotnet-format-pre-commit.cmd" to copy the pre-commit file into the GIT-REPO-LOCATION/.git/hooks directory. Verify that the file has been copied as expeted into GIT-REPO-LOCATION/.git/hooks. The cmd script assumes it is being run from the **GIT-REPO-LOCATION/format** directory
-  
+
 ```
 cd GIT-REPO-LOCATION/format
 .\dotnet-format-pre-commit.cmd
 ```
 
-After running this, when you go to commit, dotnet format should run automatically and correct any formatting that needs attention. 
+After running this, when you go to commit, dotnet format should run automatically and correct any formatting that needs attention.

src/CommonUtilities/Base32.cs

@@ -7,11 +7,13 @@ namespace CommonUtilities
 {
     public static class Base32
     {
-        private static readonly char[] Rfc4648Base32 = new[] { 'A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'I', 'J', 'K', 'L', 'M', 'N', 'O', 'P', 'Q', 'R', 'S', 'T', 'U', 'V', 'W', 'X', 'Y', 'Z', '2', '3', '4', '5', '6', '7' };
-        private const int GroupBitlength = 5;
         private const int BitsPerByte = 8;
+        private const int GroupBitlength = 5;
         private const int LargestBitPosition = GroupBitlength - 1;
 
+        private static readonly char[] Rfc4648Base32 = new[] { 'A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'I', 'J', 'K', 'L', 'M', 'N', 'O', 'P', 'Q', 'R', 'S', 'T', 'U', 'V', 'W', 'X', 'Y', 'Z', '2', '3', '4', '5', '6', '7' };
+        private static readonly string[] Rfc4648Base32Suffix = new[] { string.Empty, @"======", @"====", @"===", @"=" };
+
         /// <summary>
         /// Converts binary to Base32
         /// </summary>
@@ -23,48 +25,40 @@ public static string ConvertToBase32(byte[] bytes) // https://datatracker.ietf.o
             => new string(new BitArray(bytes).Cast<bool>()
 
                     // Reverse each byte's bits to convert the stream from LSB to MSB
-                    .ConvertGroup(BitsPerByte,
+                    .ConvertByBatch(BitsPerByte,
                         (bit, _) => bit,
                         (bits) => bits.Reverse())
                     .SelectMany(b => b)
 
                     // Convert each 5-bit group in the stream into its final character
-                    .ConvertGroup(GroupBitlength,
+                    .ConvertByBatch(GroupBitlength,
                         (bit, index) => bit ? 1 << LargestBitPosition - index : 0,
                         (values) => Rfc4648Base32[values.Sum()])
                     .ToArray())
 
                 // Append suffix
-                + (bytes.Length % GroupBitlength) switch
-                {
-                    0 => string.Empty,
-                    1 => @"======",
-                    2 => @"====",
-                    3 => @"===",
-                    4 => @"=",
-                    _ => throw new InvalidOperationException(), // Keeps the compiler happy.
-                };
+                + Rfc4648Base32Suffix[bytes.Length % GroupBitlength];
 
         /// <summary>
-        /// Converts each group (fixed number) of items into a new item
+        /// Converts each batch (fixed number) of items from an emumeration into a new item
         /// </summary>
         /// <typeparam name="TSource">Type of source items</typeparam>
-        /// <typeparam name="TGroupItem">Intermediate type</typeparam>
+        /// <typeparam name="TBatchMemberItem">Intermediate type</typeparam>
         /// <typeparam name="TResult">Type of the resultant items</typeparam>
         /// <param name="ts">The source enumerable of type <typeparamref name="TSource"/>.</param>
-        /// <param name="itemsPerGroup">The size of each group to create out of the entire enumeration. The last group may be smaller.</param>
-        /// <param name="groupMemberFunc">The function that prepares each <typeparamref name="TSource"/> into the value expected by <paramref name="groupResultFunc"/>. Its parameters are an item of type <typeparamref name="TSource"/> and the index of that item (starting from zero) within each group.</param>
-        /// <param name="groupResultFunc">The function that creates the <typeparamref name="TResult"/> from each group of <typeparamref name="TGroupItem"/>.</param>
-        /// <returns>An enumeration of <typeparamref name="TResult"/> from all of the groups.</returns>
-        private static IEnumerable<TResult> ConvertGroup<TSource, TGroupItem, TResult>(
+        /// <param name="itemsPerBatch">The size of each batch to create out of the entire enumeration. The last batch may be smaller.</param>
+        /// <param name="sourceToBatchMemberConverter">The function that prepares each <typeparamref name="TSource"/> into the type expected by <paramref name="batchToResultConverter"/>. Its parameters are an item of type <typeparamref name="TSource"/> and the <see cref="Int32"/> index of that item (starting from zero) within each group.</param>
+        /// <param name="batchToResultConverter">The function that creates the <typeparamref name="TResult"/> from each batch of <typeparamref name="TBatchMemberItem"/>.</param>
+        /// <returns>An enumeration of <typeparamref name="TResult"/> from all of the batches.</returns>
+        public static IEnumerable<TResult> ConvertByBatch<TSource, TBatchMemberItem, TResult>(
             this IEnumerable<TSource> ts,
-            int itemsPerGroup,
-            Func<TSource, int, TGroupItem> groupMemberFunc,
-            Func<IEnumerable<TGroupItem>, TResult> groupResultFunc)
+            int itemsPerBatch,
+            Func<TSource, int, TBatchMemberItem> sourceToBatchMemberConverter,
+            Func<IEnumerable<TBatchMemberItem>, TResult> batchToResultConverter)
             => ts
                 .Select((value, index) => (Index: index, Value: value))
-                .GroupBy(tuple => tuple.Index / itemsPerGroup)
+                .GroupBy(tuple => tuple.Index / itemsPerBatch)
                 .OrderBy(tuple => tuple.Key)
-                .Select(items => groupResultFunc(items.Select(i => groupMemberFunc(i.Value, i.Index % itemsPerGroup))));
+                .Select(items => batchToResultConverter(items.Select(i => sourceToBatchMemberConverter(i.Value, i.Index % itemsPerBatch))));
     }
 }

src/Tes.ApiClients/HttpApiClient.cs

@@ -78,7 +78,8 @@ protected HttpApiClient() { }
         /// <returns></returns>
         protected async Task<HttpResponseMessage> HttpSendRequestWithRetryPolicyAsync(
             Func<HttpRequestMessage> httpRequestFactory, CancellationToken cancellationToken, bool setAuthorizationHeader = false)
-            => await cachingRetryHandler.ExecuteWithRetryAsync(async ct =>
+        {
+            return await cachingRetryHandler.ExecuteWithRetryAsync(async ct =>
             {
                 var request = httpRequestFactory();
                 if (setAuthorizationHeader)
@@ -88,6 +89,7 @@ protected async Task<HttpResponseMessage> HttpSendRequestWithRetryPolicyAsync(
 
                 return await HttpClient.SendAsync(request, ct);
             }, cancellationToken);
+        }
 
         /// <summary>
         /// Sends a Http Get request to the URL and returns body response as string 
@@ -153,13 +155,15 @@ protected async Task<string> HttpGetRequestWithCachingAndRetryPolicyAsync(Uri re
         /// <returns></returns>
         protected async Task<string> HttpGetRequestWithRetryPolicyAsync(Uri requestUrl,
             CancellationToken cancellationToken, bool setAuthorizationHeader = false)
-            => await cachingRetryHandler.ExecuteWithRetryAsync(async ct =>
+        {
+            return await cachingRetryHandler.ExecuteWithRetryAsync(async ct =>
             {
                 //request must be recreated in every retry.
                 var httpRequest = await CreateGetHttpRequest(requestUrl, setAuthorizationHeader, ct);
 
                 return await ExecuteRequestAndReadResponseBodyAsync(httpRequest, ct);
             }, cancellationToken);
+        }
 
         /// <summary>
         /// Returns an query string key-value, with the value escaped. If the value is null or empty returns an empty string

src/Tes/Repository/TesDbContext.cs

@@ -32,7 +32,11 @@ protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder)
             {
                 // use PostgreSQL
                 optionsBuilder
-                    .UseNpgsql(ConnectionString, options => options.MaxBatchSize(1000))
+                    .UseNpgsql(ConnectionString, options =>
+                    {
+                        options.EnableRetryOnFailure();
+                        options.MaxBatchSize(1000);
+                    })
                     .UseLowerCaseNamingConvention();
             }
         }

src/Tes/Tes.csproj

@@ -6,6 +6,7 @@
 
   <ItemGroup>
     <PackageReference Include="EFCore.NamingConventions" Version="7.0.2" />
+    <PackageReference Include="LinqKit.Microsoft.EntityFrameworkCore" Version="7.1.4" />
     <PackageReference Include="Microsoft.EntityFrameworkCore" Version="7.0.3" />
     <PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="7.0.3">
       <PrivateAssets>all</PrivateAssets>
@@ -19,6 +20,11 @@
     <PackageReference Include="Npgsql.EntityFrameworkCore.PostgreSQL" Version="7.0.3" />
     <PackageReference Include="Polly" Version="7.2.3" />
     <PackageReference Include="Polly.Extensions.Http" Version="3.0.0" />
+    <PackageReference Include="System.Linq.Async" Version="6.0.1" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\CommonUtilities\CommonUtilities.csproj" />
   </ItemGroup>
 
 </Project>