fix: relax request processors load

project-dkg · Sep 23, 2024 · bd8253b · bd8253b
1 parent 66804d3
commit bd8253b
Show file tree

Hide file tree

Showing 4 changed files with 21 additions and 1 deletion.
diff --git a/dkgServiceNode/Data/DbEnsure.cs b/dkgServiceNode/Data/DbEnsure.cs
@@ -270,6 +270,8 @@ public static void Ensure(NpgsqlConnection connection, ILogger logger)
                 Ensure_0_14_0(connection);
                 logger.LogInformation("Tagging 0.14.1");
                 PuVersionUpdate("0.14.1", connection);
+                logger.LogInformation("Tagging 0.14.2");
+                PuVersionUpdate("0.14.2", connection);
             }
             catch (Exception ex)
             {

diff --git a/dkgServiceNode/Services/RequestProcessors/NodeAddProcessor.cs b/dkgServiceNode/Services/RequestProcessors/NodeAddProcessor.cs
@@ -65,10 +65,15 @@ protected override async Task ProcessRequests()
             while (!cancellationTokenSource.Token.IsCancellationRequested)
             {
                 var requests = new List<Node>();
-
+                int counter = 0;
                 while (requestQueue.TryDequeue(out var request) && requests.Count < bulkInsertLimit)
                 {
                     requests.Add(request);
+                    if (counter++ > bulkRestLimit)
+                    {
+                        counter = 0;
+                        await Task.Delay(0);
+                    }
                 }
 
                 if (requests.Count > 0)
@@ -96,6 +101,11 @@ protected override async Task ProcessRequests()
                         foreach (var request in requests)
                         {
                             ncContext!.FinalizeRegistration(request);
+                            if (counter++ > bulkRestLimit)
+                            {
+                                counter = 0;
+                                await Task.Delay(0);
+                            }
                         }
                     }
                     catch (Exception ex)

diff --git a/dkgServiceNode/Services/RequestProcessors/NrhAddProcessor.cs b/dkgServiceNode/Services/RequestProcessors/NrhAddProcessor.cs
@@ -24,6 +24,7 @@
 // POSSIBILITY OF SUCH DAMAGE.
 
 using System.Collections.Concurrent;
+using System.Diagnostics.Metrics;
 using System.Text.Json;
 using dkgServiceNode.Models;
 using Npgsql;
@@ -73,10 +74,16 @@ protected override async Task ProcessRequests()
             while (!cancellationTokenSource.Token.IsCancellationRequested)
             {
                 var requests = new List<Node>();
+                int counter = 0;
 
                 while (requestQueue.TryDequeue(out var request) && requests.Count < bulkInsertLimit)
                 {
                     requests.Add(request);
+                    if (counter++ > bulkRestLimit)
+                    {
+                        counter = 0;
+                        await Task.Delay(0);
+                    }
                 }
 
                 if (requests.Count > 0)

diff --git a/dkgServiceNode/Services/RequestProcessors/RequestProcessorBase.cs b/dkgServiceNode/Services/RequestProcessors/RequestProcessorBase.cs
@@ -34,6 +34,7 @@ public abstract class RequestProcessorBase : IDisposable
         protected readonly int databaseReconnectDelay = 3000;
         protected readonly int queueReparseDelay;
         protected readonly int bulkInsertLimit;
+        protected readonly int bulkRestLimit = 100;
 
         protected readonly CancellationTokenSource cancellationTokenSource = new();